car-server-core 0.52.1

//! The SUPERVISED AGENT PROCESS's side of the browser drawer.
//!
//! `car do --serve` builds its own [`BrowserTools`] and attaches to the daemon
//! over a persistent WebSocket session. This is what publishes that browser so
//! the Command Deck's drawer can watch and drive it, and what executes the
//! daemon's relayed calls against it. The daemon half is
//! [`crate::browser_relay`]; the wire between them is documented there.
//!
//! ```text
//!   register_conversation ──▶ browser.producer.register        (call)
//!   presentation pump ──────▶ browser.producer.presentation    (notify)
//!   frame pump ─────────────▶ browser.producer.frame           (notify)
//!   agent.browser.input   ──▶ ViewInput::apply(&BrowserTools)
//!   agent.browser.control ──▶ ViewControl::apply(&BrowserTools)  → effects
//!   agent.browser.capture ──▶ frame pump on/off
//! ```
//!
//! ## One browser, several conversations
//!
//! A supervised process builds ONE `AssistantRuntime` and multiplexes every
//! chat session through it (`AssistantService`), so it has exactly one
//! browser. Registering a conversation therefore publishes the SAME browser
//! under each conversation the process serves — two of its conversations show
//! one browser because there is one. Two conversations served by DIFFERENT
//! agents are different processes with different browsers and cannot see each
//! other at all.
//!
//! That is also why the run-end transition is reference-counted here: firing
//! `RunEnded` when one conversation's turn finishes would hand the browser
//! back to the user while ANOTHER conversation's turn was still driving it.

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

use car_browser::{FrameReceiver, ScreencastFrame};
use car_ffi_common::proxy::DaemonClient;
use serde_json::{json, Value};
use tokio::sync::{watch, Mutex};

use crate::assistant::browser_tools::{BrowserTools, SharedHostConnected};
use crate::browser_relay::{control_from_wire, effects_to_wire, input_from_wire};
use crate::browser_view::{WireFrame, WirePresentation};

/// How often a process with a watching host re-publishes its browser.
///
/// The daemon keys producers by CONNECTION, and a conversation is only
/// published while its process holds the session it registered on. Between
/// turns there is no other event to hang recovery off — the pumps are parked
/// on a watch and nothing calls into the process — so without this a transport
/// blip would leave the drawer dead until the user happened to send another
/// message.
///
/// It re-registers UNCONDITIONALLY rather than only what looks missing. The
/// old wording claimed it "does nothing unless something is actually missing",
/// and that was the bug: `registered` is cleared only by a FAILED push, while
/// `DaemonClient` re-dials silently, so a blip produces no failed write to
/// notice — and between turns there are no pushes at all, which made this
/// clock provably inert for the one case it exists for. `register_relay` is a
/// documented no-op for the same producer, so the cost is one round trip per
/// known conversation per interval.
/// How many recently-served conversations a process keeps re-publishing.
///
/// Small on purpose — see [`BrowserProducer::known`]. The drawer watches one
/// conversation at a time and falls back exactly one turn, so a handful covers
/// every republish that can actually restore a drawer; beyond that it is
/// per-sweep cost for conversations nobody is looking at.
pub const MAX_KNOWN_CONVERSATIONS: usize = 8;

pub const REPUBLISH_INTERVAL: Duration = Duration::from_secs(10);

/// How long one frame push may occupy the shared WS write before it is
/// abandoned.
///
/// Every writer on this session — `agent.chat.event` token deltas, the
/// heartbeat, relayed-call responses — shares one write mutex, so a frame push
/// that parks on a stalled socket parks all of them. Frames are the one thing
/// here that is safe to drop (the drawer's own contract is "a stale frame is
/// worthless; the next one is along shortly"), so they are the writer that
/// gives up.
const FRAME_PUSH_TIMEOUT: Duration = Duration::from_secs(5);

/// How long a relayed input may wait its turn behind an earlier one before it
/// is refused rather than applied.
///
/// Half the daemon's `RELAY_CALL_TIMEOUT`, which leaves the other half for the
/// input to actually reach the page. The point is the ceiling, not the exact
/// figure: an input that has queued this long is one the daemon is about to
/// abandon, and applying it after that would put a click or a keystroke onto a
/// page the person moved on from — with the drawer already told it failed.
const INPUT_QUEUE_TIMEOUT: Duration =
    Duration::from_secs(crate::browser_relay::RELAY_CALL_TIMEOUT.as_secs() / 2);

/// Publishes this process's browser to the daemon's drawer surface, and
/// executes what the drawer sends back.
pub struct BrowserProducer {
    client: Arc<DaemonClient>,
    tools: Arc<BrowserTools>,
    /// Conversations already published ON THE CURRENT SESSION. Cleared when a
    /// push proves the session is gone.
    registered: Mutex<HashSet<String>>,
    /// The conversations this process has RECENTLY been asked to publish,
    /// newest last, capped at [`MAX_KNOWN_CONVERSATIONS`]. Outlives a session,
    /// which is what lets [`Self::resync`] restore the drawer between turns
    /// instead of waiting for the user to send another message.
    ///
    /// Bounded, and a `VecDeque` rather than a set, because the unbounded
    /// version was a real cost: `register_conversation` is called once per
    /// TURN with that turn's session id, and `resync` walks the whole
    /// collection every 10 seconds issuing a serial round trip each. A
    /// long-lived supervised process therefore paid one WS call per turn it
    /// had ever served, every 10s, forever — and past the point where N × RTT
    /// exceeded the interval the pump simply ran continuously. It was also
    /// self-defeating: re-registering a long-dead conversation recreates its
    /// view on the daemon, so the daemon's own eviction could never reclaim
    /// anything this process had once served.
    ///
    /// The cap is what recovery actually needs. The drawer watches one
    /// conversation at a time and its candidate chain reaches back exactly one
    /// turn, so the recent tail is the only part a republish can help.
    known: Mutex<VecDeque<String>>,
    /// Serializes relayed input application — see the `Relayed::Input` arm.
    input_order: Mutex<()>,
    /// Chat turns in flight. The browser is shared, so the run-end transition
    /// belongs to the LAST turn to finish, not the first.
    turns: AtomicUsize,
    /// Whether a drawer is watching. Frames — and the CDP screencast behind
    /// them — cost nothing until one is.
    capture: watch::Sender<bool>,
    /// How many `agent.browser.capture` pushes this process has handled.
    ///
    /// The register ack carries the daemon's authoritative capture state, but
    /// the registration is a ROUND TRIP: a watcher arriving or leaving while it
    /// is in flight sends its own `agent.browser.capture`, which the process
    /// applies immediately — and then the ack lands carrying the older answer
    /// and overwrites it. The two sides then disagree until the next change:
    /// screencasting for nobody, or a frameless drawer. Snapshotted before the
    /// call and compared in `apply_register_ack`, which skips the capture field
    /// (only the capture field) when a newer push has already been applied.
    capture_pushes: AtomicUsize,
    /// Task 7's signal: is a CarHost host-client connected to the daemon,
    /// per the last `browser.producer.register` acknowledgment this process
    /// received? This process has no direct read of the daemon's session
    /// set (that is the in-daemon path's shortcut — see
    /// `BrowserTools::daemon_host_connectivity`), so it learns this from the
    /// registration round trip instead. Installed on `self.tools` as a
    /// [`SharedHostConnected`] probe, so `ensure_launched`'s headless
    /// decision and `browser_await_signin`'s host-gone check read the SAME
    /// flag this updates.
    ///
    /// Freshness bound, stated plainly: refreshed whenever
    /// [`Self::register_conversation`] reaches the daemon, which is once per
    /// turn (that method no longer short-circuits — see its doc comment) plus
    /// any re-registration `resync` retries. The daemon also PUSHES
    /// `agent.browser.host_connected` on every transition, so this is the
    /// slower of two paths, not the only one.
    host_connected: Arc<AtomicBool>,
}

impl BrowserProducer {
    /// Build the producer, arm the daemon's reverse-call handlers, and start
    /// the pumps. Costs nothing until a conversation is registered and a
    /// drawer subscribes: `BrowserTools` launches Chromium lazily and both
    /// pumps park on a watch.
    pub fn install(client: Arc<DaemonClient>, tools: Arc<BrowserTools>) -> Arc<Self> {
        let (capture, capture_rx) = watch::channel(false);
        let host_connected = Arc::new(AtomicBool::new(false));
        // Task 7: install the probe BEFORE this producer (or anything that
        // could reach a browse tool call) exists, so `BrowserTools` always
        // has a live read of the last-known host-connected state, however
        // stale — never a launch-time "no probe installed" default.
        tools.set_host_connectivity(Arc::new(SharedHostConnected(Arc::clone(&host_connected))));
        let producer = Arc::new(Self {
            client,
            tools: Arc::clone(&tools),
            input_order: Mutex::new(()),
            registered: Mutex::new(HashSet::new()),
            known: Mutex::new(VecDeque::new()),
            turns: AtomicUsize::new(0),
            capture,
            capture_pushes: AtomicUsize::new(0),
            host_connected,
        });

        let client = Arc::clone(&producer.client);
        for (method, kind) in [
            ("agent.browser.input", Relayed::Input),
            ("agent.browser.control", Relayed::Control),
            ("agent.browser.capture", Relayed::Capture),
            ("agent.browser.host_connected", Relayed::HostConnected),
        ] {
            let producer = Arc::clone(&producer);
            client.register_handler(method, move |params: Value| {
                let producer = Arc::clone(&producer);
                async move { producer.handle(kind, &params).await }
            });
        }

        tokio::spawn(presentation_pump(
            Arc::downgrade(&producer),
            Arc::clone(&tools),
        ));
        tokio::spawn(frame_pump(Arc::downgrade(&producer), tools, capture_rx));
        tokio::spawn(republish_pump(Arc::downgrade(&producer)));
        producer
    }

    /// Publish this process's browser for one chat session, so
    /// `browser.view.subscribe { conversation_id }` reaches it.
    ///
    /// Called at the start of every turn — the daemon needs a LIVE chat
    /// session to accept the claim, which is what stops one agent publishing
    /// into another's drawer. Best-effort by design: a chat turn must never
    /// fail because the drawer plumbing did.
    ///
    /// **Deliberately not short-circuited on `registered`.** The daemon keys
    /// producers by CONNECTION, and `DaemonClient` reconnects transparently
    /// on its next call — `car do --serve`'s 30s `models.list` heartbeat
    /// exists to be exactly that trigger — so a socket that drops between
    /// turns is normally replaced with no failed push anywhere to notice it.
    /// `registered` therefore still named a conversation the daemon had
    /// forgotten: `register_conversation` returned early, `resync`'s
    /// `known.difference(&registered)` was empty, every later
    /// `browser.producer.frame`/`presentation` was dropped for having no
    /// producer on this connection, and the drawer stayed dead for the life
    /// of the process. Re-registering costs one round trip per turn and the
    /// daemon's `register_relay` is a genuine no-op for the same producer, so
    /// the self-healing version is also the cheap one.
    pub async fn register_conversation(&self, conversation_id: &str) {
        {
            let mut known = self.known.lock().await;
            // Newest last, no duplicates, oldest evicted past the cap.
            if let Some(at) = known.iter().position(|id| id == conversation_id) {
                known.remove(at);
            }
            known.push_back(conversation_id.to_string());
            while known.len() > MAX_KNOWN_CONVERSATIONS {
                known.pop_front();
            }
        }
        let presentation = self.presentation().await;
        self.publish_conversation(conversation_id, presentation)
            .await;
    }

    /// The round trip half of [`Self::register_conversation`], with the
    /// presentation supplied — so a sweep over N conversations sharing one
    /// browser builds it once instead of N times.
    async fn publish_conversation(&self, conversation_id: &str, presentation: WirePresentation) {
        // A completed round trip is the ONLY evidence this conversation is
        // published on the connection we hold right now, so the claim is
        // dropped while it is being re-established: a re-registration that
        // fails then leaves the honest answer behind, which is exactly what
        // `resync` retries from.
        self.registered.lock().await.remove(conversation_id);
        // Read before the round trip: anything the daemon pushes DURING it is
        // newer than the answer this call comes back with.
        let capture_pushes = self.capture_pushes.load(Ordering::Acquire);
        let result = self
            .client
            .call(
                "browser.producer.register",
                json!({
                    "conversation_id": conversation_id,
                    "presentation": presentation,
                }),
            )
            .await;
        match result {
            Ok(ack) => {
                self.registered
                    .lock()
                    .await
                    .insert(conversation_id.to_string());
                self.apply_register_ack(&ack, capture_pushes);
            }
            Err(e) => tracing::debug!(
                conversation_id,
                error = %e,
                "browser producer: could not publish this browser to the drawer"
            ),
        }
    }

    /// Refresh Task 7's host-connected flag from a successful
    /// `browser.producer.register` acknowledgment. Split out from
    /// [`Self::register_conversation`] so this logic is testable without a
    /// live daemon round trip. Missing/unparseable `host_connected` (an
    /// older daemon, or a malformed ack) leaves the flag unchanged rather
    /// than guessing.
    fn apply_register_ack(&self, ack: &Value, capture_pushes_before: usize) {
        if let Some(host_connected) = ack.get("host_connected").and_then(Value::as_bool) {
            self.host_connected.store(host_connected, Ordering::Release);
        }
        // The daemon is authoritative about whether anything is watching. This
        // process's `capture` watch survives a reconnect while the daemon's
        // per-producer signal does not, so a process that was capturing when
        // its session dropped would otherwise keep screencasting and pushing
        // JPEGs under a fresh producer that never sends it anything. A `watch`
        // collapses a no-op, so agreeing costs nothing.
        //
        // Unless a push overtook it. `agent.browser.capture` is the same
        // daemon's newer word on the same question, and this ack was computed
        // before it — applying it would move the process BACKWARDS onto a
        // superseded answer, with nothing to correct it until the next watcher
        // change. Only the capture field is skipped; `host_connected` has its
        // own push path and its own ordering discipline on the daemon side.
        if self.capture_pushes.load(Ordering::Acquire) != capture_pushes_before {
            return;
        }
        if let Some(capture) = ack.get("capture").and_then(Value::as_bool) {
            let _ = self.capture.send(capture);
        }
    }

    /// Re-publish everything this process has ever published.
    ///
    /// The recovery path for a daemon session that went away between turns.
    ///
    /// It re-registers EVERYTHING known, not `known.difference(&registered)`.
    /// `registered` cannot answer "is this still published": it is cleared
    /// only by a failed push, and `DaemonClient` re-dials transparently on
    /// its next call, so an ordinary WS blip produces no failed write
    /// anywhere. Between turns there are no pushes at all — the pumps are
    /// parked on a watch — so the difference set was guaranteed empty exactly
    /// when this clock was supposed to be doing its job, and the drawer
    /// stayed dead until the user sent another message. Re-registering is
    /// self-healing for the same reason `register_conversation` no longer
    /// short-circuits: the daemon's `register_relay` is a genuine no-op for
    /// the same producer.
    ///
    /// Two bounds keep that from costing what it did when the detector was
    /// first fixed and the cost model was not revisited with it:
    ///
    /// - **Nothing runs while no host is connected.** The drawer is the only
    ///   consumer of a registration, so with no CarHost attached the whole
    ///   sweep is pure waste. The next `register_conversation` on a real turn
    ///   re-establishes everything anyway.
    /// - **The presentation is built ONCE per sweep, not once per
    ///   conversation.** Building it costs a CDP round trip per open tab, and
    ///   every conversation this process serves shares one browser (the
    ///   documented v1 property), so N conversations were paying N times for
    ///   the identical answer every 10 seconds.
    pub async fn resync(&self) {
        // The host-connected gate does NOT apply when this process holds no
        // registration at all — which is precisely the state that needs
        // healing, and the one the gate could never leave.
        //
        // `host_connected` has two writers: a successful `register` ack, and
        // the daemon's `agent.browser.host_connected` push. The push iterates
        // the daemon's producer map, which only a registration populates — so
        // a process whose producer went away can never be told the host
        // arrived, and its cached `false` gated the sweep that would have
        // re-registered it. The two signals blocked each other, forever.
        if !self.host_connected.load(Ordering::Acquire) && !self.registered.lock().await.is_empty()
        {
            return;
        }
        let known = self.known.lock().await.clone();
        if known.is_empty() {
            return;
        }
        let presentation = self.presentation().await;
        for conversation_id in &known {
            self.publish_conversation(conversation_id, presentation.clone())
                .await;
        }
    }

    /// A chat turn started.
    pub fn note_turn_started(&self) {
        self.turns.fetch_add(1, Ordering::AcqRel);
    }

    /// A chat turn finished. When it was the last one in flight, the run has
    /// ended: no ceremony, the user's browser again. The browser itself stays
    /// — this process outlives the run — so the page the agent left is still
    /// there and every control accepts input immediately.
    pub async fn note_turn_ended(&self) {
        let previous = self
            .turns
            .fetch_update(Ordering::AcqRel, Ordering::Acquire, |n| {
                Some(n.saturating_sub(1))
            })
            .unwrap_or(0);
        if previous != 1 {
            return;
        }
        let (presentation, _effects) = crate::browser_view::ViewControl::RunEnded
            .apply(&self.tools)
            .await;
        self.push_presentation(WirePresentation::from(&presentation))
            .await;
    }

    /// The current presentation in wire form.
    async fn presentation(&self) -> WirePresentation {
        WirePresentation::from(&self.tools.presentation().await)
    }

    /// Has anything been published yet? The pumps stay silent until then —
    /// building a presentation costs a CDP round trip per open tab, and a
    /// process nobody has opened a drawer on should not pay it.
    async fn is_registered(&self) -> bool {
        !self.registered.lock().await.is_empty()
    }

    async fn push_presentation(&self, presentation: WirePresentation) {
        let sent = self
            .client
            .notify(
                "browser.producer.presentation",
                json!({ "presentation": presentation }),
            )
            .await;
        self.note_push(sent).await;
    }

    /// Push one frame, giving up rather than parking forever on a stalled
    /// socket — see [`FRAME_PUSH_TIMEOUT`]. A dropped frame is already the
    /// documented policy one layer up (the daemon's fan-out drops for a slow
    /// subscriber), so it is the honest thing to do here too.
    async fn push_frame(&self, frame: WireFrame) {
        let push = self
            .client
            .notify("browser.producer.frame", json!({ "frame": frame }));
        match tokio::time::timeout(FRAME_PUSH_TIMEOUT, push).await {
            Ok(sent) => self.note_push(sent).await,
            Err(_) => tracing::debug!(
                "browser producer: dropped a frame the daemon session did not accept in time"
            ),
        }
    }

    /// A push that could not be delivered means the session this browser was
    /// published on is gone. `DaemonClient` reconnects on the next call, but
    /// the daemon keys producers by CONNECTION — the reconnected session is a
    /// different one, holding no producer at all — so what was registered on
    /// the old session is not registered any more. Forgetting it here is what
    /// makes the next turn re-publish instead of leaving the drawer dead for
    /// the life of the process.
    async fn note_push(&self, sent: Result<(), String>) {
        let Err(error) = sent else {
            return;
        };
        let dropped = {
            let mut registered = self.registered.lock().await;
            let dropped = !registered.is_empty();
            registered.clear();
            dropped
        };
        if dropped {
            tracing::debug!(
                %error,
                "browser producer: lost the daemon session; will republish on the next turn"
            );
        }
    }

    /// One relayed call from the daemon. Split out from the handler
    /// registration so tests can drive the real thing without a daemon.
    pub async fn handle(&self, kind: Relayed, params: &Value) -> Result<Value, String> {
        match kind {
            Relayed::Input => {
                let input = input_from_wire(params)?;
                // Serialized. The daemon dispatches each relayed call on its
                // own spawned task, so two keystrokes in flight together
                // reach here in scheduler order, and `apply` is several
                // awaits — two presses could interleave and land out of
                // order in the page. One lock makes the AGENT side apply them
                // one at a time; it does not fix daemon-side send order (that
                // needs a per-producer queue, and is noted as such), but it
                // removes the half this process owns.
                //
                // Bounded, because the daemon's side of this call is: past
                // `RELAY_CALL_TIMEOUT` it has already answered the drawer with
                // an error and moved on, and applying the input anyway makes
                // this at-least-once — a click landing on a page the person
                // left thirty seconds ago. Refusing fast instead makes the
                // error the drawer already showed the truth.
                let _serial = match tokio::time::timeout(
                    INPUT_QUEUE_TIMEOUT,
                    self.input_order.lock(),
                )
                .await
                {
                    Ok(guard) => guard,
                    Err(_) => {
                        return Err(
                            "the browser is still applying earlier input — the drawer gave up \
                             waiting for this one"
                                .to_string(),
                        )
                    }
                };
                // The SAME code the in-daemon path runs, error strings
                // included — see `ViewInput::apply`.
                let opened = input.apply(&self.tools).await?;
                let mut out = json!({ "ok": true });
                if let (Some(out), Some(tab_id)) = (out.as_object_mut(), opened) {
                    out.insert("tab_id".to_string(), json!(tab_id));
                }
                Ok(out)
            }
            Relayed::HostConnected => {
                // The reason this is pushed rather than polled: the flag
                // decides whether `browser_await_signin` returns the
                // "open the CAR app" result or silently waits out its whole
                // timeout in front of a drawer that is not there. Refreshing
                // it only on a registration ack meant a host that
                // disconnected mid-run was never noticed — live-observed as a
                // 60s silent poll naming a drawer nobody could see.
                let connected = params
                    .get("connected")
                    .and_then(Value::as_bool)
                    .ok_or("agent.browser.host_connected requires { connected }")?;
                self.host_connected.store(connected, Ordering::Release);
                Ok(json!({ "ok": true }))
            }
            Relayed::Control => {
                let action = params
                    .get("action")
                    .and_then(Value::as_str)
                    .ok_or("agent.browser.control requires { action }")?;
                let (presentation, effects) = control_from_wire(action)?.apply(&self.tools).await;
                // The effects have to cross back: the daemon owns the clock
                // the grace period runs on.
                Ok(json!({
                    "presentation": WirePresentation::from(&presentation),
                    "effects": effects_to_wire(&effects),
                }))
            }
            Relayed::Capture => {
                let enabled = params
                    .get("enabled")
                    .and_then(Value::as_bool)
                    .ok_or("agent.browser.capture requires { enabled }")?;
                // Counted BEFORE the send, so a registration whose ack is still
                // in flight can tell that a newer answer has landed — see
                // `apply_register_ack`.
                self.capture_pushes.fetch_add(1, Ordering::AcqRel);
                let _ = self.capture.send(enabled);
                Ok(json!({ "ok": true }))
            }
        }
    }
}

/// Pairs one already-started chat turn with its guaranteed `note_turn_ended()`.
///
/// `car do --serve`'s `agent.chat` handler calls `note_turn_started()`
/// synchronously (so the count is right before the RPC even acks), then
/// spawns a task to run the turn, which used to call `note_turn_ended()` as
/// its own trailing statement. A panic anywhere in the turn between those
/// two calls skipped `note_turn_ended()` entirely: `turns` never returned to
/// 0, `RunEnded` never fired, and the browser stayed agent-owned — no
/// strip, no take-control, no user-control blackout able to engage — for
/// the rest of the process's life, since nothing else ever calls it.
///
/// Owning a `TurnGuard` for the scope of that spawned task fixes it:
/// `Drop` runs on ANY exit from the scope that holds it, including a
/// panicking unwind, not just the happy path. `Drop` itself is synchronous
/// and `note_turn_ended` is async, so the actual call is a best-effort
/// spawned task — the same tolerance `note_turn_ended`'s own callers already
/// accept (the drawer plumbing must never be allowed to fail a chat turn).
pub struct TurnGuard {
    producer: Arc<BrowserProducer>,
}

impl TurnGuard {
    /// Wrap `producer` so `note_turn_ended()` is guaranteed to run once this
    /// guard drops. Callers still call `note_turn_started()` themselves,
    /// synchronously, before constructing this guard — tying the increment
    /// to the guard's own construction would move it into the spawned
    /// task's scheduling, which cannot guarantee it lands before the
    /// caller's RPC ack the way the direct call does.
    pub fn new(producer: Arc<BrowserProducer>) -> Self {
        Self { producer }
    }
}

impl Drop for TurnGuard {
    fn drop(&mut self) {
        let producer = Arc::clone(&self.producer);
        tokio::spawn(async move {
            producer.note_turn_ended().await;
        });
    }
}

/// Which relayed call is being handled. An enum so the three handler
/// registrations share one body and a fourth cannot be added without a
/// compile error here.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Relayed {
    Input,
    Control,
    Capture,
    /// The daemon telling this process that a CarHost host-client connected
    /// or disconnected. Pushed on every transition, so the sign-in gate does
    /// not have to wait for the next registration ack to learn the drawer is
    /// gone.
    HostConnected,
}

/// Push a presentation whenever this process's browser changes: a control
/// transition, a browser launching, a tab opening or navigating.
///
/// The same signals [`crate::browser_view`]'s in-daemon streamer watches — the
/// difference is only that this one ends in a WS notification instead of a
/// fan-out.
async fn presentation_pump(producer: Weak<BrowserProducer>, tools: Arc<BrowserTools>) {
    let mut changes = tools.subscribe_changes();
    let mut tabs = tools.subscribe_tabs().await;
    let mut last: Option<WirePresentation> = None;

    loop {
        tokio::select! {
            changed = changes.changed() => {
                if changed.is_err() {
                    return;
                }
                // A browser may have just launched — bind the tab watch if we
                // could not before.
                if tabs.is_none() {
                    tabs = tools.subscribe_tabs().await;
                }
            }
            // Only polled once a browser exists; rebound above when one appears.
            tabs_changed = async {
                match tabs.as_mut() {
                    Some(rx) => rx.changed().await.is_ok(),
                    None => std::future::pending().await,
                }
            } => {
                if !tabs_changed {
                    tabs = None;
                    continue;
                }
            }
        }

        let Some(producer) = producer.upgrade() else {
            return;
        };
        if !producer.is_registered().await {
            continue;
        }
        let presentation = producer.presentation().await;
        if last.as_ref() == Some(&presentation) {
            continue;
        }
        last = Some(presentation.clone());
        producer.push_presentation(presentation).await;
    }
}

/// Take the newest frame available, discarding everything queued behind it.
///
/// The screencast queue is bounded (`car_browser::FRAME_CHANNEL_CAP`) and
/// drops on full, so this is no longer about preventing unbounded growth —
/// it is about what the drawer SEES. Each queued slot
/// holds a full-viewport JPEG, so anything that slows the WS write — a busy
/// socket, a drawer on a slow link — turns into a growing delay before the
/// picture on screen matches the page, and then into dropped frames once the
/// buffer fills. Both are answered by the same rule: a stale frame is
/// worthless, so keep only
/// the last one. This is the same policy the daemon's fan-out already applies
/// to a slow subscriber, applied one hop earlier.
fn coalesce_newest(first: ScreencastFrame, frames: &mut FrameReceiver) -> ScreencastFrame {
    let mut newest = first;
    let mut dropped = 0usize;
    while let Ok(next) = frames.try_recv() {
        newest = next;
        dropped += 1;
    }
    if dropped > 0 {
        tracing::debug!(
            dropped,
            "browser producer: coalesced a screencast backlog to the newest frame"
        );
    }
    newest
}

/// Forward screencast frames while a drawer is watching, and nothing at all
/// when none is: subscribing is what starts CDP capture, so an unwatched
/// browser never pays for it.
async fn frame_pump(
    producer: Weak<BrowserProducer>,
    tools: Arc<BrowserTools>,
    mut capture: watch::Receiver<bool>,
) {
    loop {
        while !*capture.borrow_and_update() {
            if capture.changed().await.is_err() {
                return;
            }
        }
        let (mut frames, _epoch) = tools.subscribe_frames().await;
        loop {
            tokio::select! {
                changed = capture.changed() => {
                    if changed.is_err() {
                        return;
                    }
                    if !*capture.borrow_and_update() {
                        break;
                    }
                }
                frame = frames.recv() => {
                    match frame {
                        Some(frame) => {
                            let Some(producer) = producer.upgrade() else {
                                return;
                            };
                            // Collapse whatever piled up behind this one first:
                            // every queued slot is a
                            // full-viewport JPEG, so a drawer that fell behind
                            // must not turn into a growing backlog of pictures
                            // of pages nobody is looking at any more.
                            let newest = coalesce_newest(frame, &mut frames);
                            producer.push_frame(WireFrame::from(newest)).await;
                        }
                        // The fan-out dropped our consumer (a pump generation
                        // ended). Re-register rather than spinning on a closed
                        // channel — same recovery the in-daemon streamer makes.
                        None => {
                            let (rx, _epoch) = tools.subscribe_frames().await;
                            frames = rx;
                        }
                    }
                }
            }
        }
        drop(frames);
        // Stop paying for the screencast the moment the drawer closes.
        tools.release_frames().await;
    }
}

/// Restore anything this process published on a daemon session that has since
/// gone away. See [`REPUBLISH_INTERVAL`] for why an idle process needs a clock
/// to do it at all.
async fn republish_pump(producer: Weak<BrowserProducer>) {
    loop {
        tokio::time::sleep(REPUBLISH_INTERVAL).await;
        let Some(producer) = producer.upgrade() else {
            return;
        };
        producer.resync().await;
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::assistant::browser_control::ControlOwner;
    use crate::browser_view::WireOwner;

    /// A producer whose daemon is not there. Every push fails silently, which
    /// is exactly what production does — so the relayed-call behaviour under
    /// test is the real thing, not a stub.
    fn offline_producer() -> Arc<BrowserProducer> {
        BrowserProducer::install(
            DaemonClient::with_url("ws://127.0.0.1:1"),
            Arc::new(BrowserTools::new(std::env::temp_dir())),
        )
    }

    #[tokio::test]
    async fn a_relayed_input_runs_against_this_process_s_own_browser() {
        let producer = offline_producer();

        // No browser has launched, so this fails where the in-daemon path
        // fails, with the same words: the relay adds no behaviour of its own.
        let err = producer
            .handle(
                Relayed::Input,
                &json!({ "op": "click", "x": 1.0, "y": 2.0 }),
            )
            .await
            .unwrap_err();
        assert!(err.contains("no browser is running"), "got: {err}");
    }

    #[tokio::test]
    async fn a_malformed_relayed_input_is_refused_before_the_browser() {
        let producer = offline_producer();
        for (params, needle) in [
            (json!({ "op": "click", "x": 1.0 }), "requires { x, y }"),
            (json!({ "op": "navigate" }), "requires { url }"),
            (
                json!({ "op": "keypress", "modifiers": ["hyper"] }),
                "unknown modifier",
            ),
            (
                json!({ "op": "read_dom" }),
                "unknown agent.browser.input op",
            ),
            (json!({}), "requires { op }"),
        ] {
            let err = producer.handle(Relayed::Input, &params).await.unwrap_err();
            assert!(err.contains(needle), "{params}: got {err}");
        }
    }

    #[tokio::test]
    async fn a_relayed_control_transition_drives_the_reducer_and_reports_it_back() {
        let producer = offline_producer();
        producer.tools.attach_agent_for_test().await;

        let out = producer
            .handle(Relayed::Control, &json!({ "action": "take_control" }))
            .await
            .unwrap();
        assert_eq!(out["presentation"]["owner"], "user");
        assert_eq!(out["presentation"]["blackout_active"], true);
        assert_eq!(out["effects"], json!([]));

        // The disconnect transition is the one whose effect the daemon needs:
        // it owns the grace-period clock.
        let out = producer
            .handle(
                Relayed::Control,
                &json!({ "action": "holder_disconnected" }),
            )
            .await
            .unwrap();
        assert_eq!(out["effects"][0]["effect"], "start_grace_period");

        let out = producer
            .handle(Relayed::Control, &json!({ "action": "hand_back" }))
            .await
            .unwrap();
        assert_eq!(out["presentation"]["owner"], "agent");
        assert_eq!(out["presentation"]["blackout_active"], false);
    }

    #[tokio::test]
    async fn an_unknown_control_action_is_refused_rather_than_guessed() {
        let producer = offline_producer();
        let err = producer
            .handle(Relayed::Control, &json!({ "action": "seize" }))
            .await
            .unwrap_err();
        assert!(
            err.contains("unknown agent.browser.control action"),
            "got: {err}"
        );
    }

    #[tokio::test]
    async fn capture_toggles_the_frame_pump_and_nothing_else() {
        let producer = offline_producer();
        assert!(!*producer.capture.borrow());

        producer
            .handle(Relayed::Capture, &json!({ "enabled": true }))
            .await
            .unwrap();
        assert!(*producer.capture.borrow());

        producer
            .handle(Relayed::Capture, &json!({ "enabled": false }))
            .await
            .unwrap();
        assert!(!*producer.capture.borrow());

        let err = producer
            .handle(Relayed::Capture, &json!({}))
            .await
            .unwrap_err();
        assert!(err.contains("requires { enabled }"), "got: {err}");
    }

    /// The process has ONE browser and several conversations. Ending one
    /// conversation's turn while another is still running must not hand the
    /// browser back underneath the agent that is still driving it.
    #[tokio::test]
    async fn the_run_ends_when_the_last_turn_does_not_the_first() {
        let producer = offline_producer();
        producer.tools.attach_agent_for_test().await;

        producer.note_turn_started();
        producer.note_turn_started();

        producer.note_turn_ended().await;
        assert_eq!(
            producer.tools.control_status().await.owner,
            ControlOwner::Agent,
            "one conversation finished; the other is still driving this browser"
        );

        producer.note_turn_ended().await;
        assert_eq!(
            producer.tools.control_status().await.owner,
            ControlOwner::NoAgent,
            "the last turn ended: the user's browser again, no ceremony"
        );
    }

    /// `TurnGuard`'s happy path: dropping it at normal scope exit ends the
    /// turn exactly once, same as calling `note_turn_ended()` directly.
    #[tokio::test]
    async fn turn_guard_ends_the_turn_when_it_drops_normally() {
        let producer = offline_producer();
        producer.tools.attach_agent_for_test().await;
        producer.note_turn_started();

        {
            let _guard = TurnGuard::new(Arc::clone(&producer));
        }
        // The guard's Drop spawns the actual `note_turn_ended()` call
        // (Drop is sync, the call is async) — poll briefly rather than
        // assuming it already landed the instant Drop returns.
        for _ in 0..50 {
            if producer.tools.control_status().await.owner == ControlOwner::NoAgent {
                break;
            }
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
        assert_eq!(
            producer.tools.control_status().await.owner,
            ControlOwner::NoAgent,
            "the guard must end the turn on normal drop"
        );
    }

    /// The regression this exists for: a panic between `note_turn_started()`
    /// and the trailing `note_turn_ended()` call used to skip the latter
    /// entirely, leaving `turns` stuck above 0 and the browser agent-owned
    /// for the rest of the process's life. A `TurnGuard` held across the
    /// panic must still end the turn, because `Drop` runs during unwind.
    #[tokio::test]
    async fn turn_guard_ends_the_turn_even_if_the_task_panics() {
        let producer = offline_producer();
        producer.tools.attach_agent_for_test().await;
        producer.note_turn_started();

        let p = Arc::clone(&producer);
        let handle = tokio::spawn(async move {
            let _guard = TurnGuard::new(Arc::clone(&p));
            panic!("simulated turn panic between start and the old trailing note_turn_ended()");
        });
        let outcome = handle.await;
        assert!(outcome.is_err(), "the spawned task should have panicked");

        for _ in 0..50 {
            if producer.tools.control_status().await.owner == ControlOwner::NoAgent {
                break;
            }
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
        assert_eq!(
            producer.tools.control_status().await.owner,
            ControlOwner::NoAgent,
            "the guard's Drop must still end the turn after a panic — this is the fix"
        );
    }

    /// The browser is NOT torn down at run end — this process outlives the
    /// run, so the page the agent left is still there and drivable.
    #[tokio::test]
    async fn the_browser_outlives_the_run_that_used_it() {
        let producer = offline_producer();
        producer.tools.attach_agent_for_test().await;
        producer.note_turn_started();
        producer.note_turn_ended().await;

        let presentation = producer.presentation().await;
        assert_eq!(presentation.owner, WireOwner::None);
        assert!(!presentation.blackout_active);
        // Still the same browser, still driven through the same relay: the
        // input reaches it and fails only because this test never launched
        // Chromium.
        let err = producer
            .handle(
                Relayed::Input,
                &json!({ "op": "click", "x": 1.0, "y": 2.0 }),
            )
            .await
            .unwrap_err();
        assert!(err.contains("no browser is running"), "got: {err}");
    }

    #[tokio::test]
    async fn a_failed_registration_is_retried_on_the_next_turn() {
        let producer = offline_producer();
        // The daemon is unreachable, so this cannot succeed...
        producer.register_conversation("conv-1").await;
        assert!(
            !producer.is_registered().await,
            "a registration that never landed must not be remembered as landed"
        );
        // ...and the pumps stay silent rather than paying for a presentation
        // nobody can receive.
        assert!(!producer.is_registered().await);
        // Task 7: a registration that never landed must not update the
        // host-connected flag either — the daemon told this process nothing.
        assert!(!producer.tools.host_connected_for_test().await);
    }

    // ---- Task 7: the supervised process learns "is a host connected" from
    // `browser.producer.register`'s acknowledgment. See that method's own
    // doc comment for the freshness bound this implies.

    #[tokio::test]
    async fn a_successful_registration_ack_updates_the_shared_host_connected_flag() {
        let producer = offline_producer();
        assert!(!producer.tools.host_connected_for_test().await);

        producer.apply_register_ack(
            &json!({ "ok": true, "host_connected": true }),
            producer.capture_pushes.load(Ordering::Acquire),
        );
        assert!(
            producer.tools.host_connected_for_test().await,
            "BrowserTools reads the SAME flag this ack updates"
        );

        producer.apply_register_ack(
            &json!({ "ok": true, "host_connected": false }),
            producer.capture_pushes.load(Ordering::Acquire),
        );
        assert!(!producer.tools.host_connected_for_test().await);
    }

    /// FAIL 6: the cached flag was refreshed ONLY by a registration ack, so a
    /// host that disconnected mid-run was never noticed — `browser_await_signin`
    /// polled its full 60s and then named a drawer that was not there. The
    /// daemon now pushes every transition; this is the receiving half.
    #[tokio::test]
    async fn a_pushed_host_transition_updates_the_flag_live() {
        let producer = offline_producer();
        producer.apply_register_ack(
            &json!({ "ok": true, "host_connected": true }),
            producer.capture_pushes.load(Ordering::Acquire),
        );
        assert!(producer.tools.host_connected_for_test().await);

        // The host disconnects. No registration happens in between — that is
        // the whole point.
        let out = producer
            .handle(Relayed::HostConnected, &json!({ "connected": false }))
            .await
            .expect("the process accepts the transition");
        assert_eq!(out["ok"], true);
        assert!(
            !producer.tools.host_connected_for_test().await,
            "the sign-in gate must see the host is gone without waiting for a re-registration"
        );

        // And back again when it returns.
        producer
            .handle(Relayed::HostConnected, &json!({ "connected": true }))
            .await
            .unwrap();
        assert!(producer.tools.host_connected_for_test().await);
    }

    #[tokio::test]
    async fn a_host_transition_without_the_field_is_a_clean_error_not_a_silent_flip() {
        let producer = offline_producer();
        producer.apply_register_ack(
            &json!({ "ok": true, "host_connected": true }),
            producer.capture_pushes.load(Ordering::Acquire),
        );
        let err = producer
            .handle(Relayed::HostConnected, &json!({}))
            .await
            .unwrap_err();
        assert!(err.contains("requires { connected }"), "got: {err}");
        assert!(
            producer.tools.host_connected_for_test().await,
            "a malformed push must not change what the process believes"
        );
    }

    #[tokio::test]
    async fn an_ack_with_no_host_connected_field_leaves_the_flag_unchanged() {
        // An older daemon, or a malformed ack — don't guess.
        let producer = offline_producer();
        producer.apply_register_ack(
            &json!({ "ok": true, "host_connected": true }),
            producer.capture_pushes.load(Ordering::Acquire),
        );
        assert!(producer.tools.host_connected_for_test().await);

        producer.apply_register_ack(
            &json!({ "ok": true }),
            producer.capture_pushes.load(Ordering::Acquire),
        );
        assert!(
            producer.tools.host_connected_for_test().await,
            "a missing field must not silently reset a known-true flag to false"
        );
    }

    /// The daemon keys producers by CONNECTION, so a reconnect leaves nothing
    /// registered on the daemon's side. A process that kept believing it was
    /// published would leave the drawer dead for the rest of its life.
    #[tokio::test]
    async fn losing_the_daemon_session_makes_the_next_turn_republish() {
        let producer = offline_producer();
        producer
            .registered
            .lock()
            .await
            .insert("conv-1".to_string());
        assert!(producer.is_registered().await);

        // A push that cannot be delivered is how this process learns.
        producer
            .push_presentation(producer.presentation().await)
            .await;

        assert!(
            !producer.is_registered().await,
            "what was published on a session that is gone is not published any more"
        );
    }

    /// The other way a session goes away: no failed push at all.
    /// `DaemonClient` reconnects transparently on its next call, and
    /// `car do --serve`'s 30s heartbeat is precisely that trigger, so a WS
    /// that drops between turns (both pumps parked on a watch, pushing
    /// nothing) is replaced with nothing anywhere noticing. The daemon keys
    /// producers by CONNECTION, so the reconnected session holds no producer
    /// — while this process, short-circuiting on `registered`, never
    /// re-registered and `resync`'s `known.difference(&registered)` stayed
    /// empty. Every frame dropped, forever, on a live socket.
    #[tokio::test]
    async fn a_turn_re_registers_even_when_this_process_believes_it_already_did() {
        let producer = offline_producer();
        producer
            .registered
            .lock()
            .await
            .insert("conv-1".to_string());
        assert!(producer.is_registered().await);

        // The next turn's registration. Offline, so the round trip fails —
        // which is the observable proof it was ATTEMPTED rather than skipped.
        producer.register_conversation("conv-1").await;

        assert!(
            !producer.is_registered().await,
            "an already-registered conversation must still go to the wire, and a round \
             trip that fails must not leave this process believing it is published"
        );
        assert!(
            producer.known.lock().await.iter().any(|id| id == "conv-1"),
            "and resync must have something to retry from"
        );
    }

    /// FINDING 4, process half. Nothing calls into an idle process between
    /// turns, so recovery has to be something it does on its own: what it has
    /// EVER published outlives the session it published on, and `resync` — the
    /// republish clock's only job — retries exactly what is missing.
    #[tokio::test]
    async fn what_this_process_published_outlives_the_session_it_published_on() {
        let producer = offline_producer();
        producer.register_conversation("conv-1").await;
        // Offline, so nothing landed — but the process remembers being asked.
        assert!(!producer.is_registered().await);
        assert!(producer.known.lock().await.iter().any(|id| id == "conv-1"));

        // The republish clock's work: everything known and not currently
        // published is retried, with no user turn involved.
        producer.resync().await;
        assert!(producer.known.lock().await.iter().any(|id| id == "conv-1"));

        assert_eq!(producer.known.lock().await.len(), 1);
    }

    /// The republish clock was inert for the case it exists for. `registered`
    /// is cleared only by a FAILED push, `DaemonClient` re-dials silently, and
    /// between turns there are no pushes at all — so a transport blip left the
    /// process believing it was still published, `known.difference(&registered)`
    /// computed empty, and the drawer stayed dead until the user sent another
    /// message.
    #[tokio::test]
    async fn resync_republishes_even_when_this_process_believes_it_is_still_published() {
        let producer = offline_producer();
        // A host is attached, so a drawer could be watching — see the
        // host-connectivity bound in `resync`.
        producer.host_connected.store(true, Ordering::Release);
        producer.known.lock().await.push_back("conv-1".to_string());
        producer
            .registered
            .lock()
            .await
            .insert("conv-1".to_string());
        assert!(producer.is_registered().await);

        // Offline, so the round trip fails — which is the observable proof it
        // was ATTEMPTED rather than skipped as "nothing missing".
        producer.resync().await;

        assert!(
            !producer.is_registered().await,
            "resync must re-register everything known, not only what it thinks is missing"
        );
        assert!(producer.known.lock().await.iter().any(|id| id == "conv-1"));
    }

    /// `known` grew per TURN and was never pruned, and `resync` walks the
    /// whole thing every 10s issuing a serial round trip each — so a
    /// long-lived supervised process paid one WS call per turn it had ever
    /// served, every sweep, forever. It was self-defeating too: re-registering
    /// a long-dead conversation recreates its view on the daemon, so the
    /// daemon's own eviction could never reclaim anything.
    #[tokio::test]
    async fn the_republish_set_keeps_only_the_recent_tail() {
        let producer = offline_producer();
        for turn in 0..(MAX_KNOWN_CONVERSATIONS + 5) {
            producer
                .known
                .lock()
                .await
                .push_back(format!("turn-{turn}"));
        }
        // Re-registering trims to the cap and keeps the NEWEST.
        producer.register_conversation("newest").await;

        let known = producer.known.lock().await;
        assert_eq!(known.len(), MAX_KNOWN_CONVERSATIONS);
        assert_eq!(
            known.back().map(String::as_str),
            Some("newest"),
            "the most recent turn is what a republish can actually restore"
        );
        assert!(
            !known.iter().any(|id| id == "turn-0"),
            "and the oldest is evicted rather than re-registered forever"
        );
    }

    /// Re-asking for a conversation already in the set MOVES it to the newest
    /// end rather than duplicating it — otherwise the cap would evict live
    /// conversations to make room for repeats of one.
    #[tokio::test]
    async fn re_registering_refreshes_recency_instead_of_duplicating() {
        let producer = offline_producer();
        producer.register_conversation("a").await;
        producer.register_conversation("b").await;
        producer.register_conversation("a").await;

        let known = producer.known.lock().await;
        assert_eq!(known.len(), 2);
        assert_eq!(known.back().map(String::as_str), Some("a"));
    }

    /// The republish clock exists to keep the DRAWER alive, and the drawer
    /// needs a connected host. Re-registering every known conversation every
    /// 10s — each one a CDP tab enumeration — while nobody can possibly be
    /// watching is pure waste; the next real turn re-establishes everything
    /// anyway.
    #[tokio::test]
    async fn resync_does_nothing_while_no_host_is_connected() {
        let producer = offline_producer();
        producer.known.lock().await.push_back("conv-1".to_string());
        producer
            .registered
            .lock()
            .await
            .insert("conv-1".to_string());

        producer.resync().await;

        assert!(
            producer.is_registered().await,
            "nothing was attempted, so nothing was un-registered"
        );
    }

    /// FINDING 3. Every queued slot is a
    /// full-viewport JPEG, so a stalled write must not turn into a backlog:
    /// what goes out is the CURRENT page, and everything queued behind it is
    /// released rather than held.
    #[tokio::test]
    async fn a_stalled_push_keeps_only_the_newest_frame_and_releases_the_rest() {
        let (tx, mut rx) =
            tokio::sync::mpsc::channel::<ScreencastFrame>(car_browser::FRAME_CHANNEL_CAP);
        let frame = |byte: u8| ScreencastFrame {
            jpeg: vec![byte; 64].into(),
            viewport: car_browser::Viewport {
                width: 1920,
                height: 1080,
                device_pixel_ratio: 1.0,
            },
            captured_at: byte as f64,
        };
        // A backlog piles up while a push is parked on the shared WS write.
        for byte in 1..=9u8 {
            tx.try_send(frame(byte)).unwrap();
        }
        let first = rx.recv().await.unwrap();

        let newest = coalesce_newest(first, &mut rx);

        assert_eq!(
            newest.captured_at, 9.0,
            "the drawer gets the current page, not the oldest picture of it"
        );
        assert!(
            rx.try_recv().is_err(),
            "and the eight stale JPEGs are released, not held"
        );

        // With nothing queued it is exactly the frame it was handed.
        tx.try_send(frame(11)).unwrap();
        let only = rx.recv().await.unwrap();
        assert_eq!(coalesce_newest(only, &mut rx).captured_at, 11.0);
    }

    /// FINDING 3, the other half: a frame push must RETURN — it holds the
    /// shared WS write, and every other writer on this session (chat tokens,
    /// the heartbeat, relayed-call replies) queues behind it. This test
    /// terminating is the assertion; a push that parked would hang it.
    ///
    /// Honest coverage note: this exercises the *failed* write, not a write
    /// that stalls mid-flight. There is no way to wedge a `DaemonClient`'s
    /// socket from a unit test, so [`FRAME_PUSH_TIMEOUT`] — the bound on that
    /// case — is verified by inspection.
    #[tokio::test]
    async fn a_frame_push_returns_rather_than_parking_the_shared_write() {
        let producer = offline_producer();
        producer
            .registered
            .lock()
            .await
            .insert("conv-1".to_string());

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

        assert!(
            !producer.is_registered().await,
            "the push returned AND took the lost-session path, rather than parking"
        );
    }

    /// FINDING 2. The daemon's control gate ran a WS round trip ago, against a
    /// cached presentation. The process that owns the browser re-checks
    /// immediately before injecting, so a user click cannot land while the
    /// agent is driving — which is exactly what the blackout exists to hold.
    #[tokio::test]
    async fn relayed_input_is_refused_when_this_process_s_reducer_says_the_agent_is_driving() {
        let producer = offline_producer();
        producer.tools.attach_agent_for_test().await;

        let err = producer
            .handle(
                Relayed::Input,
                &json!({ "op": "click", "x": 1.0, "y": 2.0 }),
            )
            .await
            .unwrap_err();
        assert_eq!(
            err,
            crate::browser_view::AGENT_HOLDS_CONTROL,
            "the same words the daemon's own gate uses"
        );
        assert!(
            !err.contains("no browser is running"),
            "refused BEFORE the browser, not by it"
        );

        // Take control — the transition the drawer actually sends — and the
        // same input is accepted through to the browser.
        producer
            .handle(Relayed::Control, &json!({ "action": "take_control" }))
            .await
            .unwrap();
        let err = producer
            .handle(
                Relayed::Input,
                &json!({ "op": "click", "x": 1.0, "y": 2.0 }),
            )
            .await
            .unwrap_err();
        assert!(
            err.contains("no browser is running"),
            "now it reaches the browser and fails there; got: {err}"
        );
    }

    /// And the gate opens for a pending sign-in without a Take control press —
    /// the human has to be able to type into the credential fields.
    #[tokio::test]
    async fn a_pending_signin_lets_relayed_input_through_without_taking_control() {
        use crate::assistant::browser_control::ControlEvent;
        let producer = offline_producer();
        producer.tools.attach_agent_for_test().await;
        producer
            .tools
            .apply_control_for_test(ControlEvent::SignInRequested("Sign in at x".into()))
            .await;

        let err = producer
            .handle(Relayed::Input, &json!({ "op": "type", "text": "hunter2" }))
            .await
            .unwrap_err();
        assert!(
            err.contains("no browser is running"),
            "the gate let it through to the browser; got: {err}"
        );
    }
}