car_server_core/assistant/

browser_tools.rs

//! Browser driving + session recording for the general assistant.
//!
//! Two capabilities, deliberately paired:
//!
//! - **Drive a real web app** — `car-browser`'s CDP automation (navigate,
//!   observe, click, type, scroll, wait), already used elsewhere in CAR but
//!   never reachable from the assistant.
//! - **Record what that looked like** — `browser_record_start` /
//!   `browser_record_stop` wrap CDP screencast and hand back an MP4.
//!
//! The pairing is the point. A screenshot shows a UI's final state; a recording
//! shows it BEING USED — an answer streaming in, a table populating, a menu
//! opening. Product demos, onboarding clips and training videos want the
//! second, and a deck full of stills is the compromise you make when you can't
//! record. A text-only agent can do neither.
//!
//! Same path-artifact contract as the other media providers: write a file under
//! the working root and return its PATH.
//!
//! Chromium is launched LAZILY on first use — a browser process on every
//! assistant session would be pure waste for the majority that never browse.

use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, OnceLock, Weak};
use std::time::{Duration, Instant};

use async_trait::async_trait;
use car_browser::perception::vision::VisionPerceptionPipeline;
use car_browser::{
    BrowserBackend, BrowserToolExecutor, ChromiumBackend, FrameReceiver, HistoryStep, Modifier,
    RecordingHandle, TabId, TabInfo,
};
use car_engine::ToolExecutor;
use serde_json::{json, Value};
use tokio::sync::{watch, Mutex};

use super::browser_control::{
    ControlEffect, ControlEvent, ControlOwner, Presentation, PresentationState,
};
use super::browser_stream::{FrameAudience, FrameFanout};
use crate::browser_attention::{notify_signin_transition, SignInAttention};
use crate::coder::policy::stays_under;

/// What the agent is told whenever the privacy blackout stands between it and
/// the page — whether the call never started or was cancelled mid-flight.
/// One string on purpose: the two are the same situation from the model's
/// side, and the honest answer to both is "wait, then retry".
const BLACKOUT_HOLDS_THE_PAGE: &str =
    "the user is signing in / driving the browser, so CAR cannot observe the page right now \
     — wait for them to finish, then retry";

/// Browsing reaches arbitrary network endpoints and can act on a logged-in
/// session, so it sits at the same tier as the other egress tools.
const BROWSER_TOOL_TIER: &str = "full_access";

/// Frames-per-second the recording is encoded at. The screencast itself is
/// change-driven (see `car_browser::recorder`), so this is the output rate the
/// variable-duration frames are resampled to, not a capture rate.
const OUTPUT_FPS: u32 = 24;

/// Viewport the browser launches at, and the size frames are pinned to.
const VIEWPORT_W: u32 = 1920;
const VIEWPORT_H: u32 = 1080;

/// How long a `browse_*` call waits at the tool boundary for the user to
/// hand control back before giving up. Generous on purpose — the whole
/// point of "Take control" is a human doing something (a CAPTCHA, a
/// multi-step form, picking through a confusing menu) that can genuinely
/// take a few minutes — but bounded, so a vanished user can't wedge the
/// agent's turn forever. Matches `browser_await_signin`'s own upper clamp
/// below.
const AGENT_PAUSE_TIMEOUT_SECS: u64 = 1800;

/// JPEG quality the live drawer stream captures at when no recording has
/// asked for something else. Matches `browser_record_start`'s own default,
/// so opening the drawer during a default-quality recording changes nothing.
const DEFAULT_FRAME_QUALITY: i64 = 80;

/// Task 7's sign-in fallback, for when the browser is headless (the drawer is
/// its only face) and nobody is around to see the drawer. Exact intent per
/// the plan's seam resolution: "a clear result telling the user to open
/// CarHost — not a silent hang".
const HOST_GONE_FOR_SIGNIN: &str =
    "sign-in is needed, but this browser has no visible surface right now — no CAR \
     app is connected to the daemon to show the drawer in. Open the CAR app to continue, \
     then retry browser_await_signin.";

/// Learns whether a CarHost host-client is connected to the daemon — the
/// deciding signal for the headless/headed launch default (Task 7: "default
/// the assistant browser to headless when a drawer subscriber can serve as
/// the visible surface").
///
/// Three production sources, one per place a `BrowserTools` gets built:
/// - The in-daemon assistant (`assistant_start`) — [`BrowserTools::daemon_host_connectivity`],
///   backed by `ServerState`'s live session set. Genuinely live: no caching.
/// - The standing session (opened only through `browser.view.*`, which is
///   host-management-client-only) — [`AlwaysConnected`]: the call that
///   triggers its lazy launch IS a host, by construction.
/// - The supervised agent process (`car do --serve` / CAR Chat) —
///   [`SharedHostConnected`], refreshed from `browser.producer.register`'s
///   acknowledgment (see `assistant::browser_producer`).
///
/// `None` (every existing test, and any embedder that never installs one)
/// means "no way to know" and is treated as "no host" — today's headed
/// behavior, the safe default for a pure command-line session.
#[async_trait]
pub trait HostConnectivity: Send + Sync {
    async fn any_host_connected(&self) -> bool;
}

/// The standing session's probe. `browser.view.*` — the only way to reach
/// the standing session, and what triggers its lazy launch via
/// `user_navigate` / `user_open_tab` — is host-management-client-only
/// (Task 4's authorization rule, checked before any view is resolved), so
/// whatever call is about to launch this browser is itself coming from a
/// connected host. No live check needed; the caller reaching this code IS
/// the host.
pub struct AlwaysConnected;

#[async_trait]
impl HostConnectivity for AlwaysConnected {
    async fn any_host_connected(&self) -> bool {
        true
    }
}

/// A [`HostConnectivity`] backed by a plain shared flag that some other
/// owner updates. The supervised-agent relay path (`assistant::browser_producer`)
/// is the one production user: it has no direct read of the daemon's session
/// set, so it refreshes this from `browser.producer.register`'s
/// acknowledgment instead — see that module for the freshness bound.
pub struct SharedHostConnected(pub Arc<AtomicBool>);

#[async_trait]
impl HostConnectivity for SharedHostConnected {
    async fn any_host_connected(&self) -> bool {
        self.0.load(Ordering::Acquire)
    }
}

/// A [`HostConnectivity`] backed by the daemon's own live session set — the
/// in-daemon assistant path. Holds a WEAK reference on purpose: a
/// registered `BrowserView` outlives the run that created it (Task 4's
/// ruling) and stays in `ServerState.browser_views` indefinitely, so a
/// strong handle here would be a genuine `ServerState` <-> `BrowserTools`
/// reference cycle. An upgrade failure (the daemon is gone) reads as "no
/// host" — the safe default.
struct DaemonHostConnectivity(Weak<crate::session::ServerState>);

#[async_trait]
impl HostConnectivity for DaemonHostConnectivity {
    async fn any_host_connected(&self) -> bool {
        match self.0.upgrade() {
            Some(state) => state.any_host_connected().await,
            None => false,
        }
    }
}

/// The launch-time headless/headed decision (Task 7). Dispatch on what the
/// request needs — not a runtime implementation toggle (convention #1a):
/// the model never picks a backend, `CAR_BROWSER_HEADLESS` only ever relaxes
/// or tightens the same default.
///
/// - `host_connected`: is a host-management client attached to the daemon
///   right now? When one is, the drawer is a real visible surface, so the
///   browser launches headless and the drawer becomes its only face — no
///   separate Chrome window (the outcome this task exists to deliver). When
///   none ever is (a pure CLI session), launch headed, exactly as before:
///   `browse_await_signin` needs a visible surface, and there is no drawer
///   to be one.
/// - `headless_override`: the raw `CAR_BROWSER_HEADLESS` value, if set.
///   Keeps its EXACT existing semantics — "0", unset, or empty is falsy,
///   anything else is truthy — and always wins over the host-connected
///   default, in either direction.
///
/// Pure, so the whole rule is unit-testable without a browser or a daemon.
fn decide_headless(host_connected: bool, headless_override: Option<&str>) -> bool {
    match headless_override {
        Some(v) => v != "0" && !v.is_empty(),
        None => host_connected,
    }
}

/// What `run_await_signin_with`'s timeout error tells the user to go do,
/// picked from the SAME two signals [`decide_headless`]/the host-gone gate
/// already read — checked fresh at timeout rather than assumed headed. A
/// browser launched headed always has its window (the mid-session rule:
/// headed never becomes headless); one launched headless has only the
/// drawer as a surface, and if no host is connected right now there is no
/// surface at all — naming a "browser window" in either headless case would
/// send the user looking for something that isn't there.
fn signin_timeout_hint(headless: bool, host_connected: bool) -> &'static str {
    match (headless, host_connected) {
        (false, _) => "Ask the user to complete the login in the open browser window, then retry.",
        (true, true) => {
            "Ask the user to complete the login in the CAR app's browser drawer, then retry."
        }
        (true, false) => {
            "No CAR app is connected to show the drawer right now — open the CAR app to \
             continue, then retry browser_await_signin."
        }
    }
}

/// Who is driving and whether the model may look — the cheap read the
/// `browser.view.*` input path makes on every call.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ControlStatus {
    pub owner: super::browser_control::ControlOwner,
    /// A sign-in request is up: the human must be able to type into the
    /// page even though the agent still nominally owns the browser.
    pub signin_pending: bool,
    pub blackout_active: bool,
}

pub struct BrowserTools {
    root: PathBuf,
    /// Which profile policy this browser launches against. Passed as a
    /// launch OPTION — never via the process-global env var, which would
    /// change every other browser in the daemon (`browser.run`'s included).
    profile: BrowserProfile,
    /// Lazily launched. `None` until the first browse call.
    inner: Arc<Mutex<Option<Session>>>,
    recording: Arc<Mutex<Option<RecordingHandle>>>,
    /// A recording the RUN ended under: capture already stopped and the concat
    /// manifest already written, so its frames end at run-end.
    ///
    /// The privacy rule this exists for: nothing tears the browser down at run
    /// end (`browser_producer.rs`, `mcp_assistant.rs` both say so explicitly)
    /// and `browser_view.rs` keeps admitting the person's input once
    /// `owner == NoAgent` — so the page kept changing, the change-driven
    /// screencast kept emitting, and a recording the agent never stopped kept
    /// writing the person's OWN post-run browsing to disk. Run end stops it.
    ///
    /// Held rather than discarded so the model still gets the recording it
    /// asked for: a later `browser_record_stop` encodes exactly these frames.
    finished_recording: Arc<Mutex<Option<car_browser::recorder::Recording>>>,
    /// The live-stream quality to restore when the current recording stops —
    /// see `run_record_start_with`.
    recording_restore_quality: Arc<Mutex<Option<i64>>>,
    /// The control-ownership + presentation-state core (see
    /// `assistant::browser_control`). Who's driving, the agent's current
    /// action label, the pending sign-in strip, and the tab list the drawer
    /// shows — independent of whether a browser has even launched yet.
    presentation: Arc<Mutex<PresentationState>>,
    /// Bumped every time something a drawer subscriber would want to see
    /// changed — a control-state transition, or a browser launching. The
    /// `browser.view.*` fan-out awaits this instead of polling
    /// `presentation()`, whose tab refresh costs a CDP round trip per tab.
    changes: watch::Sender<u64>,
    /// Live screencast fan-out: the drawer and disk recording share ONE CDP
    /// screencast, with the privacy blackout applied per consumer (R3).
    frames: Arc<FrameFanout>,
    /// Task 7's launch-time signal — see [`HostConnectivity`]. Installed
    /// once by whichever producer builds this `BrowserTools`, always before
    /// any browse call can possibly fire. `None` (every existing test, and
    /// any caller that never installs one) means "no way to know" — treated
    /// as "no host", i.e. today's headed default.
    host_connectivity: OnceLock<Arc<dyn HostConnectivity>>,
    /// Whether THIS instance's browser actually launched headless — set
    /// once, inside `ensure_launched`, and fixed for this instance's
    /// lifetime (mid-session rule: a browser that launched headed stays
    /// headed; one that launched headless never becomes a visible window).
    /// `OnceLock<bool>` rather than reading `inner`'s launch state so the
    /// sign-in host-gone check ([`Self::effective_headless`]) needs no lock
    /// and no live browser to answer "not yet launched".
    launched_headless: OnceLock<bool>,
    /// Where a pending-sign-in transition becomes an operator-facing
    /// `host.event` — see [`crate::browser_attention`]. Same injection shape
    /// as [`HostConnectivity`] above, and for the same reason: this is a
    /// daemon capability a `BrowserTools` cannot build for itself, installed
    /// by whichever producer constructs it. `None` (every test, `car do`, any
    /// embedder with no daemon) means "no way to tell anyone", which is
    /// today's behavior.
    ///
    /// Carries the view key with it because the conversation id is known at
    /// the CONSTRUCTION site, not in here: a `BrowserTools` has no idea which
    /// conversation it was registered under.
    signin_attention: OnceLock<SignInAttentionBinding>,
}

/// A [`SignInAttention`] plus the conversation key to report it under, plus
/// the last state the operator was actually told about.
struct SignInAttentionBinding {
    attention: Arc<dyn SignInAttention>,
    /// `None` is the standing session — see
    /// [`crate::browser_attention::SignInAttention`].
    conversation_id: Option<String>,
    /// The pending sign-in as last ANNOUNCED, which is not the same thing as
    /// the reducer's current state.
    ///
    /// This is what makes the announcement order safe. `apply_control`
    /// deliberately releases the presentation lock before broadcasting (the
    /// broadcast awaits every `host.subscribe` subscriber's channel, and
    /// holding that lock across it would freeze every drawer input), so two
    /// concurrent transitions — a `HandBack` from the drawer racing the
    /// sign-in tool's own `SignInRequested` — could otherwise reach the host
    /// in the opposite order to the one the reducer landed them in, leaving a
    /// badge asserting a wait that already ended. Comparing against the LIVE
    /// state under this lock instead of against a value captured earlier
    /// means whoever announces last announces the truth.
    announced: Mutex<Option<String>>,
    /// Serializes the announcements themselves, so [`Self::announced`] never
    /// has to be.
    ///
    /// `record_event` awaits every `host.subscribe` socket in turn, each
    /// bounded at 10s, and `apply_control` — the single mutator, so every
    /// drawer input, Take control and Hand back — runs through the notify
    /// path. Holding `announced` across that broadcast made N backpressured
    /// host sockets a N x 10s stall on the next input, whether or not it had
    /// anything to announce.
    ///
    /// Taken while `announced` is still held and released only after the
    /// broadcast, so the ordering guarantee above survives intact: two
    /// concurrent transitions still reach the host in the order they were
    /// decided. Non-transitions — the overwhelming majority — compare and
    /// return without ever touching it.
    ///
    /// **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<()>,
}

struct Session {
    backend: Arc<ChromiumBackend>,
    /// Behind an `Arc` so a browse call can clone the handle out of
    /// `inner`'s guard and DROP the lock before awaiting the tool — see the
    /// `browse_` arm of [`BrowserTools::execute`]. Holding `inner` across
    /// `execute()` froze every drawer input and every presentation snapshot
    /// for the whole agent action.
    exec: Arc<BrowserToolExecutor>,
}

/// Which Chromium profile policy a `BrowserTools` launches against.
///
/// Persistent purposes use separate directories; coder sessions are ephemeral.
/// Chromium allows exactly ONE live
/// instance per profile directory, and an agent browsing while the person
/// uses the drawer's standing session is an ordinary situation, not an edge
/// case. Sharing one directory made the second launch die on SingletonLock;
/// nothing in the design requires the two to share cookies.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum BrowserProfile {
    /// `~/.car/browser-profile` — the agent's browser, locked decision 7's
    /// one persistent profile, unchanged.
    Agent,
    /// `~/.car/browser-profile-user` — the drawer's standing session. Its
    /// sign-ins persist across launches independently of any agent's.
    StandingSession,
    /// Fresh backend-owned profile; no cookies shared with another session.
    Isolated,
}

impl BrowserProfile {
    /// What to pass as `LaunchOptions.profile_dir`.
    ///
    /// `None` when the operator has set `CAR_BROWSER_PROFILE_DIR`, so
    /// car-browser's own resolution order (`explicit → env → ephemeral`)
    /// falls through to the env var and the knob relocates this browser, as
    /// it did before per-purpose directories existed. The derived default
    /// below is only the value used when the operator has expressed no
    /// preference.
    ///
    /// CAR code still never WRITES that variable — the invariant the
    /// grep test pins — it just stops shadowing it.
    fn launch_dir(self) -> Option<PathBuf> {
        if self == Self::Isolated {
            return None;
        }
        if std::env::var("CAR_BROWSER_PROFILE_DIR")
            .ok()
            .is_some_and(|v| !v.is_empty())
        {
            return None;
        }
        self.dir()
    }

    /// The directory, or `None` when neither `CAR_HOME` nor a home directory
    /// resolves — in which case the browser launches on car-browser's
    /// throwaway per-instance profile, exactly as it did before persistence
    /// existed.
    fn dir(self) -> Option<PathBuf> {
        let leaf = match self {
            BrowserProfile::Agent => "browser-profile",
            BrowserProfile::StandingSession => "browser-profile-user",
            BrowserProfile::Isolated => return None,
        };
        car_home::root().map(|root| root.join(leaf))
    }
}

impl BrowserTools {
    /// The agent's browser — the persistent profile an assistant run drives.
    pub fn new(root: PathBuf) -> Self {
        Self::with_profile(root, BrowserProfile::Agent)
    }

    /// A coder session owns a fresh, ephemeral profile for its browser lifetime.
    pub(crate) fn isolated(root: PathBuf) -> Self {
        Self::with_profile(root, BrowserProfile::Isolated)
    }

    /// The drawer's standing session, on its own persistent profile.
    pub fn standing_session(root: PathBuf) -> Self {
        Self::with_profile(root, BrowserProfile::StandingSession)
    }

    fn with_profile(root: PathBuf, profile: BrowserProfile) -> Self {
        Self {
            root,
            profile,
            inner: Arc::new(Mutex::new(None)),
            recording: Arc::new(Mutex::new(None)),
            finished_recording: Arc::new(Mutex::new(None)),
            recording_restore_quality: Arc::new(Mutex::new(None)),
            presentation: Arc::new(Mutex::new(PresentationState::new())),
            changes: watch::channel(0).0,
            frames: Arc::new(FrameFanout::new(
                VIEWPORT_W,
                VIEWPORT_H,
                DEFAULT_FRAME_QUALITY,
            )),
            host_connectivity: OnceLock::new(),
            launched_headless: OnceLock::new(),
            signin_attention: OnceLock::new(),
        }
    }

    /// Build a [`HostConnectivity`] backed by `state`'s live session set —
    /// the in-daemon producer's probe (`assistant_start` and anything else
    /// that builds an `AssistantRuntime` inside the daemon). Install it via
    /// [`Self::set_host_connectivity`] right after `build_assistant_runtime`
    /// returns, before the run can reach its first browse call.
    pub fn daemon_host_connectivity(
        state: &Arc<crate::session::ServerState>,
    ) -> Arc<dyn HostConnectivity> {
        Arc::new(DaemonHostConnectivity(Arc::downgrade(state)))
    }

    /// Install the probe used to decide headless vs headed at launch, and to
    /// notice a host disconnecting mid-run for the sign-in fallback.
    /// `OnceLock`-backed: every production caller installs this exactly
    /// once, before any browse call can possibly fire, so a second call
    /// (unreachable in practice) is a silent no-op rather than a panic.
    pub fn set_host_connectivity(&self, probe: Arc<dyn HostConnectivity>) {
        let _ = self.host_connectivity.set(probe);
    }

    /// Install the operator-attention sink for this browser, under the
    /// conversation key it is registered as (`None` = the standing session).
    ///
    /// `OnceLock`-backed for the same reason as
    /// [`Self::set_host_connectivity`]: every production caller installs this
    /// exactly once, before the run can reach a browse call, so a second call
    /// is a silent no-op rather than a panic.
    pub fn set_signin_attention(
        &self,
        attention: Arc<dyn SignInAttention>,
        conversation_id: Option<String>,
    ) {
        let _ = self.signin_attention.set(SignInAttentionBinding {
            attention,
            conversation_id,
            announced: Mutex::new(None),
            announce_order: Mutex::new(()),
        });
    }

    /// Is a host connected RIGHT NOW, per whatever probe was installed?
    /// `false` when none was (a pure CLI embedding, or any test).
    async fn any_host_connected(&self) -> bool {
        match self.host_connectivity.get() {
            Some(probe) => probe.any_host_connected().await,
            None => false,
        }
    }

    /// Whether the browser this call is about to (or already does) serve is
    /// headless: the ACTUAL launch decision once one has been made, or the
    /// decision `ensure_launched` would make right now otherwise. Reading
    /// this needs no lock and no live Chromium — the sign-in host-gone check
    /// calls it before ever touching a session.
    fn effective_headless(&self, host_connected: bool) -> bool {
        match self.launched_headless.get() {
            Some(&headless) => headless,
            None => decide_headless(
                host_connected,
                std::env::var("CAR_BROWSER_HEADLESS").ok().as_deref(),
            ),
        }
    }

    /// Advertised whenever a Chromium is plausibly launchable. Deliberately not
    /// probing by launching a browser at prompt-build time — that would pay the
    /// cost the lazy launch exists to avoid, on every session.
    pub fn tool_defs(&self) -> Vec<Value> {
        browser_tool_defs()
    }

    /// The agent's own entry point: launching here also ATTACHES the agent
    /// (ownership becomes run-scoped from this moment — see
    /// `browser_control::ControlState`).
    async fn session(&self) -> Result<Arc<ChromiumBackend>, String> {
        self.ensure_session(true).await
    }

    /// Launch (or reuse) the browser. `attach_agent` decides whether this
    /// counts as the agent taking the wheel.
    ///
    /// The user's own navigation in the drawer must NOT attach an agent — a
    /// standing session with no agent involved is "zero ceremony, no strip",
    /// and firing `AgentAttached` there would put the drawer in watch-only
    /// mode for a browser nobody is driving but the user.
    ///
    /// The attach is deliberately NOT tied to "did this call launch the
    /// browser". It fires on the first AGENT use of this `BrowserTools`,
    /// whoever launched Chromium. Tying it to the launch meant a user
    /// navigating in the drawer before the agent's first browse call
    /// permanently prevented the agent from ever attaching for that run: the
    /// agent's `session()` found a browser already there and returned early,
    /// so `owner` stayed `NoAgent` for the whole run — no strip, `take_control`
    /// a no-op, and the user-control blackout unable to engage at all.
    ///
    /// It is also NOT unconditional. `AgentAttached` sets `owner = Agent`, so
    /// firing it on every `session()` call would hand control straight back to
    /// the agent the instant the user pressed Take control — defeating the
    /// tool-boundary pause.
    async fn ensure_session(&self, attach_agent: bool) -> Result<Arc<ChromiumBackend>, String> {
        let backend = self.ensure_launched().await?;
        if attach_agent {
            self.maybe_attach_agent().await;
        }
        Ok(backend)
    }

    /// Fire `AgentAttached` when — and only when — nobody is driving:
    /// `owner == NoAgent`. See [`BrowserTools::ensure_session`] for both
    /// halves of why it is neither launch-tied nor unconditional.
    ///
    /// Derived from the control state rather than kept as a separate
    /// once-per-`BrowserTools` flag, because "one `BrowserTools` is one run"
    /// is FALSE on the shipped path: `car do --serve` hands the single
    /// process-lifetime `asm.browser` to `BrowserProducer::install`, and
    /// `TurnGuard` fires `RunEnded` on that same instance at the end of
    /// EVERY turn. A latching flag therefore attached on turn 1 and never
    /// again — so from turn 2 on, `RunEnded`'s `owner = NoAgent` was never
    /// undone and every `browse_*` call sat in `wait_for_agent_turn` for the
    /// full 30-minute bound before returning "the user is currently driving
    /// the browser", with no user anywhere near it.
    ///
    /// `NoAgent` is exactly the set of states where attaching is right: the
    /// initial state, after `RunEnded`, and after a hand-back that landed on
    /// `NoAgent` because the run had ended. `Agent` means this run already
    /// attached (a no-op), and `User` means a human is holding the wheel —
    /// never take it from them here; `wait_for_agent_turn` is what waits.
    async fn maybe_attach_agent(&self) {
        let attach = {
            let state = self.presentation.lock().await;
            state.control().owner() == ControlOwner::NoAgent
        };
        if attach {
            self.apply_control(ControlEvent::AgentAttached).await;
        }
    }

    async fn ensure_launched(&self) -> Result<Arc<ChromiumBackend>, String> {
        let mut guard = self.inner.lock().await;
        if guard.is_none() {
            // Persist cookies and localStorage across runs.
            //
            // `car-browser` defaults to a throwaway per-instance profile —
            // correct for parallel scraping, since it avoids Chromium
            // SingletonLock contention. For an ASSISTANT driving real web
            // apps it is the wrong default: every run lands on a login page,
            // so "record our app" or "check my dashboard" can never work. The
            // user signs in once and the session persists.
            //
            // Passed as a launch OPTION, never by setting
            // `CAR_BROWSER_PROFILE_DIR`. That env var is process-global: one
            // component setting it silently repointed every OTHER browser in
            // the daemon at the same directory, so opening the drawer and
            // typing one URL made every later `browser.run` die on
            // SingletonLock for the life of that daemon. CAR code never
            // writes it.
            //
            // And the option supplies only the DEFAULT: `launch_dir` yields
            // `None` when the operator has set `CAR_BROWSER_PROFILE_DIR`, so
            // car-browser's own resolution reads it and the knob still
            // relocates these browsers exactly as it did before this branch
            // existed. Passing the derived directory unconditionally shadowed
            // it — `LaunchOptions.profile_dir.or_else(env)` means an explicit
            // value wins — which silently broke a documented operator knob.
            let profile_dir = self.profile.launch_dir();
            if let Some(dir) = &profile_dir {
                let _ = std::fs::create_dir_all(dir);
            }
            // Task 7's decision rule: headless when a CarHost host-client is
            // connected to the daemon right now, so the drawer is a real
            // visible surface and can BE this browser's face — no separate
            // Chrome window. Headed when no host has ever connected (a pure
            // CLI session): a recording is FOOTAGE, it should show the app
            // as a person sees it, headless also trips bot-detection on some
            // login flows, and `browse_await_signin` needs a window to pop.
            // `CAR_BROWSER_HEADLESS` keeps overriding either way — see
            // `decide_headless`. The decision is made once, here, and fixed
            // for this instance's lifetime (`launched_headless`): a browser
            // does not change from headed to headless (or back) mid-run.
            let host_connected = self.any_host_connected().await;
            let headless = decide_headless(
                host_connected,
                std::env::var("CAR_BROWSER_HEADLESS").ok().as_deref(),
            );
            let options = car_browser::chromium::LaunchOptions {
                width: VIEWPORT_W,
                height: VIEWPORT_H,
                headless,
                extra_args: Vec::new(),
                profile_dir,
            };
            let backend = if self.profile == BrowserProfile::Isolated {
                ChromiumBackend::launch_isolated_with_options(options).await
            } else {
                ChromiumBackend::launch_with_options(options).await
            }
            .map_err(|e| format!("launch browser: {e}"))?;
            // Latched only now the launch has SUCCEEDED. Setting it first
            // meant a failed launch pinned the mode for the instance's whole
            // life — a `OnceLock`, so the retry that actually starts a
            // browser could never correct it, and every later
            // `effective_headless` answer (which decides whether
            // `browser_await_signin` reports a window or the drawer) came
            // from a browser that never existed.
            let _ = self.launched_headless.set(headless);
            let backend = Arc::new(backend);
            // A browser now exists: point the live fan-out at it (starting
            // capture if the drawer is already subscribed) and wake any
            // drawer subscriber so it re-reads the tab list.
            self.frames.bind(Arc::clone(&backend)).await;
            self.signal_change();
            *guard = Some(Session {
                backend: Arc::clone(&backend),
                exec: Arc::new(BrowserToolExecutor::new(
                    Arc::clone(&backend) as Arc<dyn car_browser::BrowserBackend>,
                    // Vision-fused perception, NOT the bare accessibility
                    // tree. The model driving the browser is text-only, so all
                    // it ever gets from browse_observe is the ui_map — and a
                    // polished custom SPA (Contrails, most React apps) exposes
                    // a poor a11y tree, so the composer and Send button simply
                    // don't appear and the agent clicks blind. VisionPerception
                    // runs OCR (Apple Vision on macOS) over the screenshot and
                    // fuses recovered labels onto the elements, so the text map
                    // actually names the controls that are on screen. Degrades
                    // to the plain tree when no OCR backend is present.
                    Arc::new(VisionPerceptionPipeline::new()),
                )),
            });
        }
        Ok(Arc::clone(&guard.as_ref().expect("just set").backend))
    }

    /// Block until the page STOPS changing — i.e. an answer has finished
    /// rendering — then return.
    ///
    /// This exists because `browse_observe` snapshots immediately: the model
    /// cannot tell "still loading" from "done", so when told to record an app
    /// answering a question it submits, waits a guessed couple of seconds, and
    /// stops recording while the app is still thinking. (Observed live: 7 of 8
    /// Contrails recordings captured the home screen because the answer hadn't
    /// arrived yet.) Polling the rendered text length until it holds steady for
    /// a few consecutive checks is a content-agnostic "it settled" signal that
    /// works without knowing anything about the site.
    async fn run_await_answer(&self, params: &Value) -> Result<Value, String> {
        self.run_await_answer_with(
            params,
            Duration::from_secs(AGENT_PAUSE_TIMEOUT_SECS),
            Duration::from_secs(1),
        )
        .await
    }

    /// `gate_timeout`/`gate_poll` parameterize the boundary wait below (see
    /// `wait_while_blackout_with`) so tests can exercise the block path
    /// without waiting `AGENT_PAUSE_TIMEOUT_SECS`.
    async fn run_await_answer_with(
        &self,
        params: &Value,
        gate_timeout: Duration,
        gate_poll: Duration,
    ) -> Result<Value, String> {
        // This tool MEASURES THE PAGE — repeatedly, for up to ten minutes,
        // returning `content_length` to the model. That is model-facing
        // observation in every sense the blackout means (R3), so it gates
        // like the other browser tools do, before touching a session. The
        // blackout predicate rather than `user_holds_control` because it
        // covers the sign-in case too, and is `false` for `NoAgent` — so a
        // first call that hasn't attached yet is never blocked.
        self.wait_while_blackout_with(gate_timeout, gate_poll)
            .await?;
        let backend = self.session().await?;
        let page = backend
            .page_handle()
            .await
            .map_err(|e| format!("no page: {e}"))?;
        let timeout = params
            .get("timeout_seconds")
            .and_then(Value::as_u64)
            // A real LLM-backed app answering a data question takes 1-2 minutes,
            // so the default is generous. Observed on Contrails: a delayed-flights
            // query sat on "Almost there…" for over 90s before the table rendered.
            .unwrap_or(150)
            .clamp(3, 600);
        // Poll interval, and how long content must hold STEADY to count as done.
        // A loading spinner animates its dots, so raw text length wobbles by a
        // few chars while "generating"; requiring a longer steady hold and a
        // tolerance band keeps that wobble from reading as "still growing"
        // forever (the bug that stranded every recording on the spinner).
        let poll = Duration::from_millis(1000);
        let steady_hold = Duration::from_secs(6);

        self.await_settled(Duration::from_secs(timeout), poll, steady_hold, || async {
            page.evaluate("document.body ? document.body.innerText.length : 0")
                .await
                .ok()
                .and_then(|v| v.into_value::<i64>().ok())
                .unwrap_or(0)
        })
        .await
    }

    /// The settle loop `browser_await_answer` runs, with the page read behind
    /// a `measure` closure.
    ///
    /// Extracted for one reason: the blackout re-check inside the loop is the
    /// half of R3 that an entry gate cannot cover — the user can press Take
    /// control, or a sign-in can appear, while this is already polling. With
    /// `measure` injected, a test can prove the page is NEVER read during a
    /// blackout without needing a live Chromium.
    ///
    /// While blacked out the loop PAUSES rather than exiting: the answer may
    /// still be rendering, and the honest thing is to keep waiting out the
    /// caller's own timeout rather than report a settled state nobody looked
    /// at. The steady clock resets, so blackout time never counts as "held
    /// steady", and the timeout path reports only values measured before the
    /// blackout began — nothing observed during it reaches the model.
    async fn await_settled<F, Fut>(
        &self,
        timeout: Duration,
        poll: Duration,
        steady_hold: Duration,
        measure: F,
    ) -> Result<Value, String>
    where
        F: Fn() -> Fut,
        Fut: std::future::Future<Output = i64>,
    {
        // Length changes at or below this are treated as noise (spinner dots,
        // a relative timestamp ticking), not real content growth.
        let noise_band: i64 = 8;

        let started = Instant::now();
        let baseline = measure().await;
        let mut last = baseline;
        let mut steady_since: Option<Instant> = None;
        let mut peak = baseline;
        while started.elapsed() < timeout {
            tokio::time::sleep(poll).await;
            // The loop already sleeps every tick, so this check is free — and
            // it is the only thing standing between the model and a page the
            // user took control of (or is typing a password into) after this
            // call started.
            if self.blackout_active().await {
                steady_since = None;
                continue;
            }
            let now = measure().await;
            peak = peak.max(now);
            if (now - last).abs() > noise_band {
                // Real change — reset the steady clock.
                steady_since = None;
                last = now;
            } else {
                // Within the noise band. Only start (or continue) counting as
                // steady once content has meaningfully GROWN past where we
                // began — otherwise a static page "settles" before the answer
                // starts, and a spinner alone never trips it.
                if peak - baseline > noise_band {
                    let since = *steady_since.get_or_insert_with(Instant::now);
                    if since.elapsed() >= steady_hold {
                        return Ok(json!({
                            "settled": true,
                            "content_length": now,
                            "waited_seconds": started.elapsed().as_secs(),
                        }));
                    }
                }
            }
        }
        Ok(json!({
            "settled": false,
            "content_length": last,
            "grew": peak - baseline > noise_band,
            "note": "timed out before the page held steady. If `grew` is true the answer was \
                     still streaming at timeout — raise timeout_seconds. If false, the action \
                     produced no visible change (the submit may not have registered).",
        }))
    }

    /// Hand the browser to the human so they can sign in, then resume.
    ///
    /// An agent driving a real web app hits auth immediately, and it cannot
    /// (and must not) type someone's password. The browser has a visible
    /// surface for the user to complete the flow on — SSO, MFA, a device
    /// prompt, whatever it is — either the drawer (browser launched
    /// headless because a host was connected) or a headed Chrome window
    /// (pure CLI session, no host ever connected) — see [`decide_headless`].
    /// If the browser is headless and nobody is around to see the drawer
    /// right now, there is no surface at all; this call fails clearly with
    /// [`HOST_GONE_FOR_SIGNIN`] instead of polling for up to `timeout`
    /// seconds against a window nobody can see (Task 7's seam resolution).
    /// This tool is the handshake otherwise: navigate, surface the ask, and
    /// block until the sign-in visibly succeeds.
    ///
    /// Completion is detected by the URL leaving the login flow, which is the
    /// one signal that works across SSO redirects without knowing anything
    /// about the site's markup.
    async fn run_await_signin(&self, params: &Value) -> Result<Value, String> {
        self.run_await_signin_with(
            params,
            Duration::from_secs(AGENT_PAUSE_TIMEOUT_SECS),
            Duration::from_secs(1),
        )
        .await
    }

    /// `gate_timeout`/`gate_poll` parameterize the boundary wait below (see
    /// `wait_while_user_holds_control_with`) so tests can exercise the
    /// block path without waiting `AGENT_PAUSE_TIMEOUT_SECS`.
    async fn run_await_signin_with(
        &self,
        params: &Value,
        gate_timeout: Duration,
        gate_poll: Duration,
    ) -> Result<Value, String> {
        // Must not navigate — or do anything else — while the user holds
        // control: this call would otherwise yank the page out from under
        // someone who just took it. Blocking here until hand-back is the
        // resolution (per the task brief's approval-race semantics): the
        // sign-in request then starts cleanly once the user is done
        // driving, exactly as if it had been called fresh. Checked BEFORE
        // `session()` — safe to do so because `user_holds_control` (unlike
        // `may_agent_act`) is false for `NoAgent`, so a call that hasn't
        // attached yet is never mistaken for "the user is driving" and
        // never blocks on its own first invocation.
        self.wait_while_user_holds_control_with(gate_timeout, gate_poll)
            .await?;

        // Task 7's host-gone fallback. A headless browser's only visible
        // surface is the drawer; if no host is connected right now there is
        // NOBODY who could complete a sign-in, so fail clearly instead of
        // polling silently for up to `timeout` seconds. Checked BEFORE
        // `session()` (so before ever touching Chromium, and testable
        // without a live one): `effective_headless` reads the ACTUAL launch
        // decision once a browser exists, or what `ensure_launched` would
        // decide right now otherwise — either way, no live check requires
        // this instance's browser to already be running.
        let host_connected = self.any_host_connected().await;
        if self.effective_headless(host_connected) && !host_connected {
            return Err(HOST_GONE_FOR_SIGNIN.to_string());
        }

        let backend = self.session().await?;
        let url_param = params
            .get("url")
            .and_then(|v| v.as_str())
            .map(str::trim)
            .filter(|s| !s.is_empty());

        // Surface the drawer's orange strip (and start the privacy
        // blackout) before navigating, not after — the request itself, not
        // just the resulting page, is what must never reach the model.
        let signin_message = match url_param {
            Some(url) => format!("Sign in at {url}"),
            None => "Sign in to continue in the browser window".to_string(),
        };
        self.apply_control(ControlEvent::SignInRequested(signin_message))
            .await;

        if let Some(url) = url_param {
            if let Err(e) = backend.navigate(url).await {
                // Don't leave the strip up for a sign-in that never starts.
                self.apply_control(ControlEvent::SignInResolved { signed_in: false })
                    .await;
                return Err(format!("navigate to {url}: {e}"));
            }
        }
        let expect = params
            .get("success_url_contains")
            .and_then(|v| v.as_str())
            .map(str::to_string);
        let timeout = params
            .get("timeout_seconds")
            .and_then(Value::as_u64)
            .unwrap_or(300)
            .clamp(10, 1800);

        let started = Instant::now();
        let mut last = String::new();
        while started.elapsed() < Duration::from_secs(timeout) {
            tokio::time::sleep(Duration::from_secs(2)).await;
            // Honest resolution: "Hand back to CAR" already resolves a
            // pending sign-in (see `ControlState::apply`'s HandBack branch)
            // without our own detection loop knowing about it directly. If
            // something else already cleared the strip, this loop is the
            // only thing left with model-facing state, so it reports the
            // same signed_in:false a genuine hand-back means.
            if self
                .presentation
                .lock()
                .await
                .control()
                .pending_signin()
                .is_none()
            {
                return Ok(json!({
                    "signed_in": false,
                    "url": last,
                    "note": "Control was handed back before sign-in was detected.",
                }));
            }
            // The same host-gone predicate the entry gate uses, re-read.
            // Checked only at entry, a host that quit mid-wait left the agent
            // polling a headless page nobody could see for up to 1800s — the
            // exact silent wait `HOST_GONE_FOR_SIGNIN` exists to replace.
            // Unlike the entry gate this must also clear the strip: by now
            // `SignInRequested` has been applied.
            let host_now = self.any_host_connected().await;
            if self.effective_headless(host_now) && !host_now {
                self.apply_control(ControlEvent::SignInResolved { signed_in: false })
                    .await;
                return Err(HOST_GONE_FOR_SIGNIN.to_string());
            }
            // The user pressed Take control while this wait was running.
            // `TakeControl` does not resolve the pending sign-in — by design,
            // the strip stays up — so without this the loop kept reading the
            // page's URL every two seconds and handed it to the model, for up
            // to 1800s, with the blackout nominally up. Worse, the no-`expect`
            // heuristic below treats any URL without login/auth in it as
            // success, so a person navigating away mid-blackout would have
            // declared the sign-in complete and handed the model that URL.
            // Keep waiting; observe nothing.
            if self
                .presentation
                .lock()
                .await
                .control()
                .user_holds_control()
            {
                continue;
            }
            let current = backend.get_current_url().unwrap_or_default();
            last = current.clone();
            let done = match &expect {
                Some(needle) => current.contains(needle.as_str()),
                // With no explicit target, treat leaving the login/auth path
                // as success — covers the common OAuth/SSO round trip.
                None => {
                    !current.is_empty()
                        && !["login", "signin", "sign-in", "auth", "oauth", "sso"]
                            .iter()
                            .any(|p| current.to_ascii_lowercase().contains(p))
                }
            };
            if done {
                self.apply_control(ControlEvent::SignInResolved { signed_in: true })
                    .await;
                return Ok(json!({
                    "signed_in": true,
                    "url": current,
                    "note": "Sign-in detected. The session persists in the browser profile, so \
                             later runs won't need this again.",
                }));
            }
        }
        // Resolved on timeout ONLY if nobody was ever here.
        //
        // Two rules that both have to hold, and this predicate is what lets
        // them. The acceptance contract: "letting the sign-in request time out
        // behaves as today — the strip clears, the agent receives the same
        // timeout result, and control returns to it." That is right when the
        // timer expired because nobody came: no strip anyone is reading, no
        // page anyone is typing into, and latching the blackout would wedge
        // every later browse call — on a one-shot `car do` run, forever, since
        // that path has no drawer, no Hand back, and no run-end signal reaching
        // this reducer at all.
        //
        // The privacy rule: a timer expiring is not evidence the person
        // finished. If somebody took control or drove the browser during this
        // window, clearing the strip lifts the blackout under them — the
        // model's page reads reopen and an in-flight recording (a
        // `FrameAudience::Model` consumer kept registered precisely so it
        // resumes) starts writing the login form to disk.
        //
        // So: engagement decides. Un-engaged, the contract's semantics
        // exactly. Engaged, the window persists until THEIR signal — hand-back,
        // run end, or the disconnect grace expiring — each of which resolves it
        // honestly as `signed_in: false`.
        let user_engaged = self.presentation.lock().await.control().user_engaged();
        if !user_engaged {
            self.apply_control(ControlEvent::SignInResolved { signed_in: false })
                .await;
        }
        let host_connected = self.any_host_connected().await;
        let headless = self.effective_headless(host_connected);
        // Naming the page is model-facing observation too. If the person is
        // still driving at the timeout, `last` is whatever was on screen
        // before they took over — stale, and not ours to report.
        //
        // Gated on ENGAGEMENT, not ownership, and on the value already computed
        // above rather than a second read. The ordinary sign-in flow never
        // involves Take control, so `user_holds_control()` is false for the
        // most common case there is — a person typing into the credential form
        // — and this branch handed the model the live page URL of exactly the
        // page the blackout above had just been kept up to hide. Same
        // asymmetry, same fix, as the `RunEnded` rule in `browser_control`.
        if user_engaged {
            return Err(format!(
                "timed out after {timeout}s waiting for sign-in — the user is still driving \
                 the browser. {}",
                signin_timeout_hint(headless, host_connected)
            ));
        }
        Err(format!(
            "timed out after {timeout}s waiting for sign-in — the browser is still at {last}. \
             {}",
            signin_timeout_hint(headless, host_connected)
        ))
    }

    async fn run_record_start(&self, params: &Value) -> Result<Value, String> {
        self.run_record_start_with(
            params,
            Duration::from_secs(AGENT_PAUSE_TIMEOUT_SECS),
            Duration::from_secs(1),
        )
        .await
    }

    /// `gate_timeout`/`gate_poll` parameterize the boundary wait below (see
    /// `wait_while_user_holds_control_with`) so tests can exercise the
    /// block path without waiting `AGENT_PAUSE_TIMEOUT_SECS`.
    async fn run_record_start_with(
        &self,
        params: &Value,
        gate_timeout: Duration,
        gate_poll: Duration,
    ) -> Result<Value, String> {
        // Same tier, same "agent-initiated mutating browser action" shape
        // as browse_*/browser_await_signin — pause here too, before ever
        // touching the session, rather than starting a recording (or
        // failing on a stale precondition) while the user is driving.
        self.wait_while_user_holds_control_with(gate_timeout, gate_poll)
            .await?;
        let mut rec = self.recording.lock().await;
        if rec.is_some() {
            return Err(
                "a recording is already in progress — call browser_record_stop first".to_string(),
            );
        }
        // A recording the previous run ended under, that nobody came back to
        // encode. Starting a new one supersedes it, so its frames are a build
        // artifact nothing will ever read — dropped here rather than left to
        // accumulate one directory per abandoned recording.
        if let Some(stale) = self.finished_recording.lock().await.take() {
            let _ = std::fs::remove_dir_all(&stale.dir);
        }
        let backend = self.session().await?;
        // Fail here, before any frame plumbing, exactly as before: a browser
        // with no page cannot be recorded.
        backend
            .page_handle()
            .await
            .map_err(|e| format!("no page to record: {e}"))?;

        let quality = params
            .get("quality")
            .and_then(Value::as_i64)
            .unwrap_or(DEFAULT_FRAME_QUALITY)
            .clamp(1, 100);
        let dir = self.root.join(format!(
            ".car-recording-{}",
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .map(|d| d.as_millis())
                .unwrap_or(0)
        ));
        // Disk recording is a consumer of the SAME screencast the drawer
        // watches (see `browser_stream`), registered as `Model` so the
        // privacy blackout keeps a user-control / sign-in window out of the
        // produced video while the drawer keeps streaming it (R3).
        // Remembered so `record_stop` can put it back. `set_quality` stores
        // into the atomic the SHARED pump reads and re-attaches it, and that
        // pump also serves every `FrameAudience::Viewer` — the person watching
        // the drawer. `quality` is a model-supplied tool argument, so
        // `browser_record_start { quality: 1 }` dropped the human's live view
        // to unreadable artifacts and left it there for the life of the
        // browser, across every later turn.
        let restore_quality = self.frames.quality();
        self.frames.set_quality(quality).await;
        let (incoming, epoch) = self.frames.subscribe(FrameAudience::Model).await;
        let started = car_browser::recorder::start_with_frames(incoming, epoch, &dir).await;
        let handle = match started {
            Ok(handle) => handle,
            Err(e) => {
                // Symmetric with `run_record_stop_with`: the consumer above is
                // registered, so a failure here owes the same release. Without
                // it a failed `start_with_frames` (an unwritable recording
                // directory is the realistic one) left a dead consumer keeping
                // the CDP screencast armed for the life of the browser. The
                // quality goes back for the same reason.
                self.release_frames().await;
                self.frames.set_quality(restore_quality).await;
                return Err(format!("start recording: {e}"));
            }
        };
        *rec = Some(handle);
        *self.recording_restore_quality.lock().await = Some(restore_quality);
        Ok(json!({
            "recording": true,
            "note": "Recording. Drive the app with the browse_* tools, then call \
                     browser_record_stop. Frames are only captured when the page \
                     CHANGES, so a static page produces nothing — make sure \
                     something actually happens on screen.",
        }))
    }

    async fn run_record_stop(&self, params: &Value) -> Result<Value, String> {
        self.run_record_stop_with(
            params,
            Duration::from_secs(AGENT_PAUSE_TIMEOUT_SECS),
            Duration::from_secs(1),
        )
        .await
    }

    /// `gate_timeout`/`gate_poll` parameterize the boundary wait below (see
    /// `wait_while_user_holds_control_with`) so tests can exercise the
    /// block path without waiting `AGENT_PAUSE_TIMEOUT_SECS`.
    /// Everything between taking the handle and having a finished recording —
    /// split out so its `?`s cannot skip the frame release its caller owes.
    #[allow(clippy::type_complexity)]
    async fn finish_recording(
        &self,
        handle: RecordingHandle,
        params: &Value,
    ) -> Result<(PathBuf, String, car_browser::recorder::Recording), String> {
        let (out, rel) = self.recording_output(params)?;
        let recording = handle
            .stop()
            .await
            .map_err(|e| format!("stop recording: {e}"))?;
        Ok((out, rel, recording))
    }

    /// Resolve and prepare `output_path`. Split out from
    /// [`Self::finish_recording`] because a recording the RUN already stopped
    /// (see [`Self::stop_recording_at_run_end`]) still needs the destination
    /// but has nothing left to stop.
    fn recording_output(&self, params: &Value) -> Result<(PathBuf, String), String> {
        let rel = params
            .get("output_path")
            .and_then(|v| v.as_str())
            .filter(|s| !s.trim().is_empty())
            .unwrap_or("assets/recording.mp4")
            .to_string();
        if !stays_under(&self.root, &rel) {
            return Err(format!("output_path '{rel}' escapes the working directory"));
        }
        let out = self.root.join(&rel);
        if let Some(parent) = out.parent() {
            std::fs::create_dir_all(parent).map_err(|e| format!("create output dir: {e}"))?;
        }
        Ok((out, rel))
    }

    /// The run ended with a recording still running: stop it here, so no frame
    /// captured after run-end can reach disk.
    ///
    /// Nothing else would. `self.recording` was taken only by `record_stop`
    /// and by `record_start`'s already-in-progress check; there is no `Drop`
    /// for `BrowserTools`, the browser is deliberately NOT torn down at run
    /// end, and the person keeps driving the page (`browser_view.rs` admits
    /// input once `owner == NoAgent`), so the change-driven screencast kept
    /// emitting into a `FrameAudience::Model` consumer that `browser_stream`
    /// deliberately keeps registered through a blackout. Their own browsing,
    /// written to the agent's recording directory, for the life of the process.
    ///
    /// The artifact is finalized rather than discarded: `handle.stop()` writes
    /// the concat manifest over the frames captured UP TO now, and the result
    /// is held so a later `browser_record_stop` still returns the model its
    /// video — one that ends where the run did.
    async fn stop_recording_at_run_end(&self) {
        let Some(handle) = self.recording.lock().await.take() else {
            return;
        };
        let finished = handle.stop().await;
        // The same two duties `record_stop` owes once the handle is out: the
        // Model consumer is this function's to release, and the drawer gets
        // its picture quality back.
        self.release_frames().await;
        if let Some(restore) = self.recording_restore_quality.lock().await.take() {
            self.frames.set_quality(restore).await;
        }
        match finished {
            Ok(recording) => {
                tracing::debug!(
                    frames = recording.frame_count,
                    "browser tools: the run ended with a recording running; stopped it there"
                );
                *self.finished_recording.lock().await = Some(recording);
            }
            // "screencast captured no frames" is the ordinary case for a page
            // that never changed — nothing to keep, and nothing to report to
            // an agent whose run is already over.
            Err(e) => tracing::debug!(
                error = %e,
                "browser tools: the recording the run ended under produced nothing"
            ),
        }
    }

    async fn run_record_stop_with(
        &self,
        params: &Value,
        gate_timeout: Duration,
        gate_poll: Duration,
    ) -> Result<Value, String> {
        self.wait_while_user_holds_control_with(gate_timeout, gate_poll)
            .await?;
        let handle = self.recording.lock().await.take();
        let Some(handle) = handle else {
            // No live recording — but possibly one the RUN stopped. Run end
            // finalizes the frames and hands the encode to whoever asks next,
            // so the model still gets its video; it simply ends where the run
            // did. Capture, the frame consumer and the drawer's quality were
            // all already settled there, so all that is left is the encode.
            let mut finished = self.finished_recording.lock().await;
            if finished.is_none() {
                return Err(
                    "no recording in progress — call browser_record_start first".to_string()
                );
            }
            let (out, rel) = self.recording_output(params)?;
            let recording = finished.take().expect("checked immediately above");
            drop(finished);
            return self.encode_recording(&out, &rel, &recording);
        };

        // The handle is OUT of `self.recording` now, so its Model consumer on
        // `self.frames` is this function's to release — on every exit, not
        // just the happy one. Three `?`s used to sit between here and the
        // release, and the third is the ordinary case: `RecordingHandle::stop`
        // errors with "screencast captured no frames" whenever the page never
        // changed, which `browser_record_start`'s own description warns the
        // model about. On a still page with no drawer subscriber no later
        // frame ever arrives to prune the dead consumer, so `consumers` stays
        // non-empty, the supervisor stays alive, and the CDP screencast stays
        // armed for the life of the browser — verbatim the regression
        // `record_stop_releases_its_frame_consumer` exists to prevent, reached
        // through the error path it did not cover.
        let recording = self.finish_recording(handle, params).await;
        // `release_frames` is a no-op for any consumer still alive (e.g. a
        // drawer subscriber watching the same browser).
        self.release_frames().await;
        // And the drawer gets its picture quality back, on every exit — the
        // recording's setting was never the human's choice.
        if let Some(restore) = self.recording_restore_quality.lock().await.take() {
            self.frames.set_quality(restore).await;
        }
        let (out, rel, recording) = recording?;
        self.encode_recording(&out, &rel, &recording)
    }

    /// Encode a finished recording's frames to `out`. Shared by the ordinary
    /// stop and by the encode owed for a recording the run already stopped.
    fn encode_recording(
        &self,
        out: &Path,
        rel: &str,
        recording: &car_browser::recorder::Recording,
    ) -> Result<Value, String> {
        // Encode from the concat manifest so each frame is held for its REAL
        // duration — the screencast is change-driven, so a fixed rate would
        // compress every pause. `-vsync cfr` resamples that variable timeline
        // to a constant output rate players handle predictably.
        let status = std::process::Command::new("ffmpeg")
            .args([
                "-nostdin", "-v", "error", "-f", "concat", "-safe", "0", "-i",
            ])
            .arg(&recording.manifest)
            .args([
                "-vsync",
                "cfr",
                "-r",
                &OUTPUT_FPS.to_string(),
                "-pix_fmt",
                "yuv420p",
                "-c:v",
                "libx264",
                "-movflags",
                "+faststart",
            ])
            .arg(out)
            .arg("-y")
            .status()
            .map_err(|e| format!("run ffmpeg (is it installed?): {e}"))?;
        if !status.success() {
            return Err(format!("ffmpeg failed encoding the recording ({status})"));
        }
        // Frames are a build artifact; the MP4 is the deliverable.
        let _ = std::fs::remove_dir_all(&recording.dir);

        let bytes = std::fs::metadata(out).map(|m| m.len()).unwrap_or(0);
        Ok(json!({
            "video_path": rel,
            "media_type": "video/mp4",
            "bytes": bytes,
            "frames": recording.frame_count,
            "duration_seconds": recording.duration_seconds,
            "note": format!(
                "Wrote a {:.1}s screen recording ({} frames) to {rel}.",
                recording.duration_seconds, recording.frame_count
            ),
        }))
    }

    /// The agent-pause-at-tool-boundary check every `browse_*` call makes
    /// before it runs: block (poll) while the user holds control, erroring
    /// only if hand-back never comes within the bounded wait — the same
    /// block-then-timeout-then-error shape `run_await_signin` above already
    /// uses, not an immediate error the instant the user takes control.
    /// When no drawer/user-control is in play (the common case today, since
    /// nothing yet calls `take_control()`) `may_agent_act()` is already
    /// true, so this returns on the very first check — zero added latency,
    /// identical behavior to before this task.
    async fn wait_for_agent_turn(&self) -> Result<(), String> {
        self.wait_for_agent_turn_with(
            Duration::from_secs(AGENT_PAUSE_TIMEOUT_SECS),
            Duration::from_secs(1),
        )
        .await
    }

    /// `timeout`/`poll` are parameterized so tests can exercise the block
    /// and unblock paths without actually waiting `AGENT_PAUSE_TIMEOUT_SECS`.
    ///
    /// Re-attempts the attach on every pass, rather than relying on the one
    /// `session()` made on the way in. A user holding control at the moment
    /// this turn started blocks that attach (never take the wheel from a
    /// person), and their hand-back lands on `NoAgent` when the previous run
    /// had already ended — so the moment the wheel frees up has to be the
    /// moment this run gets to attach, or the call waits out the full
    /// timeout against a browser nobody is driving.
    async fn wait_for_agent_turn_with(
        &self,
        timeout: Duration,
        poll: Duration,
    ) -> Result<(), String> {
        self.maybe_attach_agent().await;
        if self.presentation.lock().await.control().may_agent_act() {
            return Ok(());
        }
        let started = Instant::now();
        while started.elapsed() < timeout {
            tokio::time::sleep(poll).await;
            self.maybe_attach_agent().await;
            if self.presentation.lock().await.control().may_agent_act() {
                return Ok(());
            }
        }
        Err(
            "the user is currently driving the browser — ask them to hand back control \
             (or wait for them to finish), then retry"
                .to_string(),
        )
    }

    /// Sibling of `wait_for_agent_turn` for the three tools that need to
    /// gate BEFORE they've necessarily established a session yet —
    /// `browser_await_signin`, `browser_record_start`, `browser_record_stop`
    /// — same `full_access`, agent-initiated, mutating tier as `browse_*`,
    /// so the same tool-boundary pause and approval-race re-check apply
    /// (the brief's pause/approval-race requirements carry no `browse_`-
    /// prefix qualifier). Uses `user_holds_control` rather than
    /// `may_agent_act`: see that method's doc comment for why — in short,
    /// it must NOT block a call that hasn't attached yet, only one where
    /// the user has explicitly taken control.
    ///
    /// `timeout`/`poll` are parameterized (each of the three call sites
    /// passes `AGENT_PAUSE_TIMEOUT_SECS`/1s in production) so tests can
    /// exercise the block and unblock paths without actually waiting that
    /// long — mirrors `wait_for_agent_turn`/`wait_for_agent_turn_with`.
    async fn wait_while_user_holds_control_with(
        &self,
        timeout: Duration,
        poll: Duration,
    ) -> Result<(), String> {
        if !self
            .presentation
            .lock()
            .await
            .control()
            .user_holds_control()
        {
            return Ok(());
        }
        let started = Instant::now();
        while started.elapsed() < timeout {
            tokio::time::sleep(poll).await;
            if !self
                .presentation
                .lock()
                .await
                .control()
                .user_holds_control()
            {
                return Ok(());
            }
        }
        Err(
            "the user is currently driving the browser — ask them to hand back control \
             (or wait for them to finish), then retry"
                .to_string(),
        )
    }

    /// The blackout gate on the MODEL-facing side (controller ruling R3):
    /// while the user holds control or a sign-in is pending, no page
    /// observation may reach the model. `may_agent_act` already covers the
    /// user-control half (owner == User); this is what covers the other
    /// trigger — a pending sign-in, where the agent still nominally owns
    /// the browser but the human is typing a password into it.
    ///
    /// Blocks rather than erroring, for the same reason
    /// `wait_while_user_holds_control_with` does: a sign-in is a human-paced
    /// interruption, and the honest answer to "may I look at the page" is
    /// "not yet", not "the tool is broken".
    async fn wait_while_blackout_with(
        &self,
        timeout: Duration,
        poll: Duration,
    ) -> Result<(), String> {
        if !self.blackout_active().await {
            return Ok(());
        }
        let started = Instant::now();
        while started.elapsed() < timeout {
            tokio::time::sleep(poll).await;
            if !self.blackout_active().await {
                return Ok(());
            }
        }
        Err(BLACKOUT_HOLDS_THE_PAGE.to_string())
    }

    /// Resolves the moment a blackout becomes active, and never otherwise.
    ///
    /// The mid-flight half of the R3 gate. `wait_while_blackout_with` only
    /// answers "may this call START", and `browse_wait` is not a fast tool —
    /// its `timeout_ms` is model-supplied and unclamped, and car-browser
    /// polls the full page accessibility tree every 100ms until the deadline.
    /// A user pressing Take control ten seconds into a five-minute
    /// `browse_wait` would otherwise have the whole page they just took over
    /// read back to the model when the call returned.
    ///
    /// Watches the same `changes` signal `apply_control` bumps on every
    /// transition, so there is no poll interval to lose the edge in.
    async fn blackout_starts(&self, changes: &mut watch::Receiver<u64>) {
        loop {
            if self.blackout_active().await {
                return;
            }
            if changes.changed().await.is_err() {
                // Nothing can signal a transition any more, so a blackout can
                // never start; park rather than spin, and let the other
                // `select!` arm decide the call.
                std::future::pending::<()>().await
            }
        }
    }

    async fn blackout_active(&self) -> bool {
        self.presentation
            .lock()
            .await
            .control()
            .is_blackout_active()
    }

    /// The one mutator. Applies the event, republishes the blackout flag the
    /// frame fan-out gates on, wakes every `browser.view.*` subscriber, and —
    /// on a pending-sign-in TRANSITION only — tells the operator through the
    /// daemon's always-on `host.event` channel.
    ///
    /// This is the local path's choke point on purpose. It is the single
    /// mutator of the control reducer, so every route that can start or end a
    /// sign-in wait passes through here exactly once: the tool's own
    /// `SignInRequested`/`SignInResolved`, hand-back, run end, the disconnect
    /// grace expiring, and the host-gone bail-out. Emitting from the sign-in
    /// tool instead would cover the request and miss four of the five ways it
    /// ends — and a notification that never clears is worse than none.
    async fn apply_control(&self, event: ControlEvent) -> Vec<ControlEffect> {
        let effects = {
            let mut state = self.presentation.lock().await;
            let effects = state.apply_control(event);
            // Published UNDER the lock that produced it. Computing the flag
            // here and storing it after the lock released let two concurrent
            // transitions publish out of order: the reducer runs serialized,
            // so the STATE is right, but whichever store landed last decided
            // the gate — and a `RunEnded` computed before a `SignInRequested`
            // could store `false` after it, leaving the recording gate open
            // for the whole sign-in with the reducer insisting the blackout
            // was up. `set_blackout` is a plain `AtomicBool::store` with no
            // await, which is exactly why `FrameFanout::blackout` is an
            // atomic and not a callback (see its doc): it is safe to set
            // while this lock is held.
            self.frames
                .set_blackout(state.control().is_blackout_active());
            effects
        };
        // Outside the presentation lock: `record_event` takes the host's own
        // locks and awaits every subscriber's channel, and holding this one
        // across that would freeze every drawer input and every snapshot for
        // the duration.
        self.notify_signin_transition().await;
        self.signal_change();
        effects
    }

    /// Tell the operator if the pending sign-in has changed since the last
    /// time we told them — see [`crate::browser_attention`] for the
    /// transition rule, and `SignInAttentionBinding::announced` for why this
    /// re-reads the live state rather than taking a before/after pair from
    /// the caller.
    async fn notify_signin_transition(&self) {
        let Some(binding) = self.signin_attention.get() else {
            return;
        };
        // Taken BEFORE the presentation lock, and never the other way round:
        // `apply_control` releases that lock before calling this, so the two
        // are only ever acquired in this order.
        let mut announced = binding.announced.lock().await;
        let current = self
            .presentation
            .lock()
            .await
            .control()
            .pending_signin()
            .map(|p| p.message.clone());
        if *announced == current {
            return;
        }
        let before = std::mem::replace(&mut *announced, current.clone());
        // Queue behind any announcement still in flight BEFORE releasing the
        // decision lock — that is what preserves the order two concurrent
        // transitions were decided in — and then RELEASE it, so the broadcast
        // below does not hold every drawer input behind a backpressured host
        // socket. See `SignInAttentionBinding::announce_order`.
        let _order = binding.announce_order.lock().await;
        drop(announced);
        notify_signin_transition(
            &binding.attention,
            binding.conversation_id.as_deref(),
            before.as_deref(),
            current.as_deref(),
        )
        .await;
    }

    /// Wake anything awaiting [`Self::subscribe_changes`].
    fn signal_change(&self) {
        self.changes.send_modify(|n| *n = n.wrapping_add(1));
    }

    /// A signal that fires whenever the presentation may have changed — a
    /// control-state transition, or a browser launching. The `browser.view.*`
    /// fan-out awaits this and only then calls [`Self::presentation`], whose
    /// live tab refresh costs a CDP round trip per open tab.
    pub fn subscribe_changes(&self) -> watch::Receiver<u64> {
        self.changes.subscribe()
    }

    /// Live screencast frames for a human watching the drawer. Never gated
    /// by the blackout — that is the whole point of it (the person driving
    /// keeps seeing the page; the model does not). Capture starts on the
    /// first subscription and stops when the last one is dropped.
    pub async fn subscribe_frames(&self) -> (FrameReceiver, Instant) {
        self.frames.subscribe(FrameAudience::Viewer).await
    }

    /// Drop consumers whose receiver has gone away, stopping CDP capture if
    /// that was the last one. Called when the last drawer subscriber leaves,
    /// so a browser nobody is watching stops paying for a screencast without
    /// waiting for a frame to arrive and notice.
    pub async fn release_frames(&self) {
        self.frames.prune().await;
    }

    /// The cheap read of who is driving — no CDP, unlike
    /// [`Self::presentation`], which refreshes the tab list. This is what
    /// the input path consults on every call.
    pub async fn control_status(&self) -> ControlStatus {
        let state = self.presentation.lock().await;
        ControlStatus {
            owner: state.control().owner(),
            signin_pending: state.control().pending_signin().is_some(),
            blackout_active: state.control().is_blackout_active(),
        }
    }

    /// Tab open/close/switch/nav-state notifications from the live browser,
    /// or `None` when none has launched yet.
    pub async fn subscribe_tabs(&self) -> Option<watch::Receiver<car_browser::TabsSnapshot>> {
        self.inner
            .lock()
            .await
            .as_ref()
            .map(|s| s.backend.subscribe_tabs())
    }

    /// The presentation snapshot — tabs (live-refreshed from the browser
    /// when one exists), control owner, current action, pending sign-in,
    /// blackout, and a monotonic revision. What the `browser.view.*` RPC
    /// surface serializes.
    pub async fn presentation(&self) -> Presentation {
        self.build_presentation_snapshot().await
    }

    /// Pending prompt for the host reconnect snapshot, without the CDP tab
    /// refresh a full presentation requires.
    pub(crate) async fn pending_signin_message(&self) -> Option<String> {
        self.presentation
            .lock()
            .await
            .control()
            .pending_signin()
            .map(|pending| pending.message.clone())
    }

    /// Resolve any operator attention before an unreachable view is dropped.
    ///
    /// `GracePeriodExpired` is the reducer's existing honest "the person no
    /// longer owns this browser" ending. Running it through the one mutator
    /// keeps the blackout, presentation revision, and `host.event` twin in
    /// lockstep instead of synthesizing a notification beside the state.
    pub(crate) async fn resolve_signin_attention_on_teardown(&self) {
        if self
            .presentation
            .lock()
            .await
            .control()
            .pending_signin()
            .is_some()
        {
            self.apply_control(ControlEvent::GracePeriodExpired).await;
        }
    }

    /// User presses "Take control" in the drawer.
    pub async fn take_control(&self) -> (Presentation, Vec<ControlEffect>) {
        self.apply_and_snapshot(ControlEvent::TakeControl).await
    }

    /// User presses "Hand back to CAR" — also resolves a pending sign-in,
    /// if one is up (see `ControlState::apply`).
    pub async fn hand_back(&self) -> (Presentation, Vec<ControlEffect>) {
        self.apply_and_snapshot(ControlEvent::HandBack).await
    }

    /// The run this browser was attached to has ended: no ceremony, the
    /// user's browser again. Nothing in this crate calls this yet — the
    /// daemon owns "when does a run end" (see the R6 note in the task
    /// brief), this is the seam it hooks into.
    pub async fn note_run_ended(&self) -> (Presentation, Vec<ControlEffect>) {
        // BEFORE the transition, not after. The un-engaged `RunEnded` branch
        // clears a pending sign-in and lifts the blackout, and the blackout is
        // the only thing gating a `FrameAudience::Model` consumer — so applying
        // it first would open a window, however short, in which the recording
        // this is about to stop is capturing again.
        self.stop_recording_at_run_end().await;
        self.apply_and_snapshot(ControlEvent::RunEnded).await
    }

    /// A person drove this browser. See `ControlState::user_engaged` — this is
    /// what a sign-in timeout consults before deciding it may end their
    /// window.
    pub async fn note_user_input(&self) {
        self.apply_control(ControlEvent::UserInput).await;
    }

    /// The connection holding user control dropped. Nothing in this crate
    /// detects that condition yet (it lives at the transport layer); this
    /// is the seam a later task hooks a real disconnect signal into.
    pub async fn control_holder_disconnected(&self) -> (Presentation, Vec<ControlEffect>) {
        self.apply_and_snapshot(ControlEvent::ControlHolderDisconnected)
            .await
    }

    /// The grace period started by `control_holder_disconnected` elapsed.
    /// Nothing in this crate owns the actual clock yet (see
    /// `ControlEffect::StartGracePeriod`'s doc comment); this is the seam.
    pub async fn grace_period_expired(&self) -> (Presentation, Vec<ControlEffect>) {
        self.apply_and_snapshot(ControlEvent::GracePeriodExpired)
            .await
    }

    async fn apply_and_snapshot(&self, event: ControlEvent) -> (Presentation, Vec<ControlEffect>) {
        let effects = self.apply_control(event).await;
        (self.build_presentation_snapshot().await, effects)
    }

    // ---- User-driven input, from the drawer ------------------------------
    //
    // These are the `browser.view.*` input RPCs' only route into the
    // browser. They are deliberately narrow — navigate/click/type/keypress/
    // scroll and the three tab operations, exactly what a person does with a
    // mouse and keyboard — so this surface adds NO perception of any kind:
    // no DOM reads, no screenshots, no accessibility tree. A human watching
    // the drawer sees the page; nothing here returns page content to a
    // caller.
    //
    // None of them consults the agent-turn gate: the gate exists to stop the
    // AGENT acting while a human drives, and these calls are the human. Who
    // may call them is decided one level up, by the `browser.view.*` control
    // owner check — a caller that does not hold control never reaches here.

    /// Navigate the active tab. This is the one input that may LAUNCH the
    /// browser: the standing session starts empty and comes to life on the
    /// user's first navigation (controller ruling R6), without attaching an
    /// agent.
    pub async fn user_navigate(&self, url: &str) -> Result<(), String> {
        let url = url.trim();
        if url.is_empty() {
            return Err("navigate requires a non-empty `url`".to_string());
        }
        let backend = self.ensure_session(false).await?;
        backend
            .navigate(url)
            .await
            .map_err(|e| format!("navigate to {url}: {e}"))?;
        self.signal_change();
        Ok(())
    }

    pub async fn user_click(&self, x: f64, y: f64) -> Result<(), String> {
        let backend = self.live_backend().await?;
        backend
            .inject_click(x, y)
            .await
            .map_err(|e| format!("click: {e}"))?;
        self.signal_change();
        Ok(())
    }

    pub async fn user_type(&self, text: &str) -> Result<(), String> {
        let backend = self.live_backend().await?;
        backend
            .inject_text(text)
            .await
            .map_err(|e| format!("type: {e}"))
    }

    /// Paste `text` at the caret, replacing the selection.
    ///
    /// The host reads its OWN pasteboard and sends the string, because a
    /// synthesised ⌘V cannot work: the clipboard belongs to the browser, not
    /// the page, and CDP's injected key events have no access to it.
    pub async fn user_paste(&self, text: &str) -> Result<(), String> {
        let backend = self.live_backend().await?;
        backend
            .insert_text(text)
            .await
            .map_err(|e| format!("paste: {e}"))
    }

    pub async fn user_keypress(&self, key: &str, modifiers: &[Modifier]) -> Result<(), String> {
        let backend = self.live_backend().await?;
        backend
            .inject_keypress(key, modifiers)
            .await
            .map_err(|e| format!("keypress: {e}"))?;
        self.signal_change();
        Ok(())
    }

    pub async fn user_scroll(&self, delta_y: i32) -> Result<(), String> {
        let backend = self.live_backend().await?;
        backend
            .inject_scroll(delta_y)
            .await
            .map_err(|e| format!("scroll: {e}"))
    }

    /// The nav bar's Back button. Drives Chromium's real session history via
    /// CDP — a synthesised ⌘← keystroke does not, because the shortcut is
    /// browser chrome the input domain never reaches.
    pub async fn user_go_back(&self) -> Result<(), String> {
        self.step_history(HistoryStep::Back).await
    }

    /// The nav bar's Forward button.
    pub async fn user_go_forward(&self) -> Result<(), String> {
        self.step_history(HistoryStep::Forward).await
    }

    async fn step_history(&self, step: HistoryStep) -> Result<(), String> {
        let backend = self.live_backend().await?;
        backend
            .step_history(step)
            .await
            // The backend's message already names the direction that had
            // nowhere to go ("no page to go back to"), so it needs no prefix
            // of its own — unlike click/type/scroll, where the operation name
            // is the only thing identifying which call failed.
            .map_err(|e| e.to_string())?;
        self.signal_change();
        Ok(())
    }

    /// The nav bar's Reload button. Errors on the empty state through
    /// [`Self::live_backend`], like every other input that needs a page to
    /// act on.
    pub async fn user_reload(&self) -> Result<(), String> {
        let backend = self.live_backend().await?;
        backend.reload().await.map_err(|e| format!("reload: {e}"))?;
        self.signal_change();
        Ok(())
    }

    pub async fn user_tab_open(&self) -> Result<TabId, String> {
        let backend = self.ensure_session(false).await?;
        let id = backend
            .open_tab()
            .await
            .map_err(|e| format!("open tab: {e}"))?;
        self.signal_change();
        Ok(id)
    }

    pub async fn user_tab_close(&self, id: TabId) -> Result<(), String> {
        let backend = self.live_backend().await?;
        backend
            .close_tab(id)
            .await
            .map_err(|e| format!("close tab: {e}"))?;
        self.signal_change();
        Ok(())
    }

    pub async fn user_tab_switch(&self, id: TabId) -> Result<(), String> {
        let backend = self.live_backend().await?;
        backend
            .switch_tab(id)
            .await
            .map_err(|e| format!("switch tab: {e}"))?;
        self.signal_change();
        Ok(())
    }

    /// Resolve a wire tab id (`TabId`'s own `Display` form, e.g. `tab-3`)
    /// against the open tabs. `TabId` is deliberately opaque outside
    /// car-browser — matching on the rendered id is what keeps it that way,
    /// and it makes a stale id from a closed tab a clean error rather than
    /// an operation on some other tab.
    pub async fn resolve_tab(&self, wire_id: &str) -> Result<TabId, String> {
        let tabs: Vec<TabInfo> = self
            .live_backend()
            .await?
            .list_tabs()
            .await
            .unwrap_or_default();
        tabs.into_iter()
            .find(|t| t.id.to_string() == wire_id)
            .map(|t| t.id)
            .ok_or_else(|| format!("no open tab '{wire_id}'"))
    }

    /// The launched browser, or a clean error when there isn't one. Input
    /// that cannot sensibly launch a browser (a click needs a page to click
    /// on) fails here rather than starting Chromium for it.
    async fn live_backend(&self) -> Result<Arc<ChromiumBackend>, String> {
        self.inner
            .lock()
            .await
            .as_ref()
            .map(|s| Arc::clone(&s.backend))
            .ok_or_else(|| {
                "no browser is running for this view — navigate to a page first".to_string()
            })
    }

    /// Drive the control reducer directly. Test-only: production code
    /// reaches these transitions through the tools and the `browser.view.*`
    /// surface, never by hand.
    #[cfg(test)]
    pub async fn apply_control_for_test(&self, event: ControlEvent) -> Vec<ControlEffect> {
        self.apply_control(event).await
    }

    /// The agent's first (or Nth) tool call, minus the Chromium launch — the
    /// crate has no live-browser tests (see the browse-tool test suite's own
    /// constraint). Calls the SAME `maybe_attach_agent` production takes, so
    /// the once-per-run decision is exercised rather than re-stated.
    #[cfg(test)]
    pub async fn attach_agent_for_test(&self) {
        self.maybe_attach_agent().await;
    }

    /// Read back whatever [`HostConnectivity`] probe is installed. Test-only
    /// window into Task 7's signal, so a cross-module test (the supervised
    /// relay's own `assistant::browser_producer` suite) can assert that a
    /// registration acknowledgment actually reached this `BrowserTools`
    /// without a live daemon round trip.
    #[cfg(test)]
    pub async fn host_connected_for_test(&self) -> bool {
        self.any_host_connected().await
    }

    /// Live-refreshes the tab list from the browser (if one has launched)
    /// before building the snapshot, so a caller never sees a stale tab
    /// strip. Never holds `inner`'s lock and `presentation`'s lock at the
    /// same time — it reads the backend handle, drops that lock, awaits
    /// `list_tabs()` unlocked, then locks `presentation` only at the end.
    async fn build_presentation_snapshot(&self) -> Presentation {
        let backend = {
            self.inner
                .lock()
                .await
                .as_ref()
                .map(|s| Arc::clone(&s.backend))
        };
        let tabs = match backend {
            Some(backend) => backend.list_tabs().await.unwrap_or_default(),
            None => Vec::new(),
        };
        let mut presentation = self.presentation.lock().await;
        presentation.set_tabs(tabs);
        presentation.snapshot()
    }
}

/// A plain-words label for the drawer's action strip, derived from the tool
/// name and its most descriptive parameter — not a per-tool lookup table, so
/// a new `browse_*` tool gets a sensible label automatically. Shown only to
/// the user who is already watching the same headed browser live, so this
/// intentionally does not avoid echoing a `text` param even though a
/// `browse_type` call could in principle carry one — nothing here reaches
/// the model (that boundary is the blackout, not this label).
fn describe_browse_action(tool: &str, params: &Value) -> String {
    let verb = tool
        .strip_prefix("browse_")
        .unwrap_or(tool)
        .replace('_', " ");
    let mut label = String::new();
    let mut chars = verb.chars();
    if let Some(first) = chars.next() {
        label.extend(first.to_uppercase());
    }
    label.push_str(chars.as_str());
    for key in ["url", "text", "key", "condition", "element_id"] {
        if let Some(v) = params.get(key).and_then(Value::as_str) {
            let v = v.trim();
            if !v.is_empty() {
                label.push(' ');
                label.push_str(v);
                break;
            }
        }
    }
    label
}

pub(super) fn browser_tool_defs() -> Vec<Value> {
    let mut defs: Vec<Value> = BrowserToolExecutor::tool_schemas()
        .into_iter()
        .map(|s| {
            json!({
                "name": s.name,
                "description": s.description,
                "parameters": s.parameters,
                "mutating": !s.idempotent,
                "tier": BROWSER_TOOL_TIER,
            })
        })
        .collect();

    defs.push(json!({
        "name": "browser_await_answer",
        "description": "After you submit a question or trigger an action in a web app, call this to \
            WAIT until the response has finished rendering, before you screenshot or stop a \
            recording. It polls the page and returns once the content stops changing. Use it every \
            time between submitting and observing/recording an answer — browse_observe does NOT \
            wait, so without this you capture the page mid-load (a blank or still-thinking state) \
            instead of the actual answer.",
        "parameters": {
            "type": "object",
            "properties": {
                "timeout_seconds": {
                    "type": "integer",
                    "description": "Max seconds to wait for the page to settle (default 45)."
                }
            },
            "required": []
        },
        "mutating": false,
        "tier": BROWSER_TOOL_TIER
    }));
    defs.push(json!({
        "name": "browser_await_signin",
        "description": "Ask the USER to sign in, on whatever surface this browser has, and wait \
            until they have. Use this the moment a site needs authentication — you cannot and must \
            not type someone's credentials, but the person can complete any flow (SSO, MFA, a \
            device prompt) themselves: in the CAR app's browser drawer when the app is running, or \
            in the browser window if one is open. TELL THE USER what to sign into before calling \
            this, and say to look in the CAR app if they do not see a browser window; it blocks \
            while they do it. The session persists in the browser profile, so this is a one-time \
            cost per site rather than per run.",
        "parameters": {
            "type": "object",
            "properties": {
                "url": {
                    "type": "string",
                    "description": "Optional page to navigate to first, e.g. the app's home or login URL."
                },
                "success_url_contains": {
                    "type": "string",
                    "description": "Optional substring identifying a signed-in URL. Omit to accept any URL that no longer looks like a login/SSO page."
                },
                "timeout_seconds": {
                    "type": "integer",
                    "description": "How long to wait for the user (default 300, max 1800)."
                }
            },
            "required": []
        },
        "mutating": true,
        "tier": BROWSER_TOOL_TIER
    }));
    defs.push(json!({
        "name": "browser_record_start",
        "description": "Start RECORDING the browser session to video. Pair it with the browse_* \
            tools: start recording, drive the app (navigate, type a real question, wait for the \
            answer), then call browser_record_stop to get an MP4. Use it whenever the ASK is a \
            product demo, an onboarding or training clip, a bug repro, or release notes — anything \
            where showing the app BEING USED beats a screenshot of its final state. Frames are \
            captured only when the page actually CHANGES, so make sure something happens on \
            screen; a static page records nothing.",
        "parameters": {
            "type": "object",
            "properties": {
                "quality": {"type": "integer", "description": "JPEG quality 1-100 (default 80)."}
            },
            "required": []
        },
        "mutating": true,
        "tier": BROWSER_TOOL_TIER
    }));
    defs.push(json!({
        "name": "browser_record_stop",
        "description": "Stop the recording started by browser_record_start and write an MP4 under \
            the working directory. Returns the path plus the real duration. Requires ffmpeg.",
        "parameters": {
            "type": "object",
            "properties": {
                "output_path": {
                    "type": "string",
                    "description": "Where to write the MP4, relative to the working directory (default assets/recording.mp4)."
                }
            },
            "required": []
        },
        "mutating": true,
        "tier": BROWSER_TOOL_TIER
    }));
    defs
}

#[async_trait]
impl ToolExecutor for BrowserTools {
    async fn execute(&self, tool: &str, params: &Value) -> Result<Value, String> {
        match tool {
            "browser_await_signin" => self.run_await_signin(params).await,
            "browser_await_answer" => self.run_await_answer(params).await,
            "browser_record_start" => self.run_record_start(params).await,
            "browser_record_stop" => self.run_record_stop(params).await,
            t if t.starts_with("browse_") => {
                // Ensure the browser exists (this is also where the agent
                // formally attaches for this run — see `session()` above).
                self.session().await?;
                // Pause at the tool boundary while the user holds control.
                // Re-checked HERE, right before delegating to execution —
                // not cached from whenever this call was proposed/approved
                // — which is what closes the approval race (see
                // `ControlState::may_agent_act`'s doc comment).
                self.wait_for_agent_turn().await?;
                // The other blackout trigger: a pending sign-in, where the
                // agent still owns the browser but a human is typing a
                // password into it. Nothing the agent does here may observe
                // the page until that resolves (R3).
                self.wait_while_blackout_with(
                    Duration::from_secs(AGENT_PAUSE_TIMEOUT_SECS),
                    Duration::from_secs(1),
                )
                .await?;
                self.apply_control(ControlEvent::AgentActionStarted(describe_browse_action(
                    tool, params,
                )))
                .await;
                // Delegate to car-browser's own executor so the automation
                // semantics live in one place.
                //
                // The handle is CLONED OUT and the lock DROPPED before the
                // await. Holding `inner` across `execute()` put every
                // concurrent reader of it behind the whole agent action:
                // `live_backend()` (every drawer click/type/paste/scroll/
                // keypress/tab op) and `build_presentation_snapshot()`
                // (`presentation`, `take_control`, `hand_back`,
                // `note_run_ended`). A model-supplied, unclamped
                // `browse_wait { timeout_ms }` therefore froze Take control
                // and every keystroke for as long as the model asked for.
                let exec = {
                    let guard = self.inner.lock().await;
                    Arc::clone(&guard.as_ref().ok_or("browser session unavailable")?.exec)
                };
                // Raced against a blackout STARTING, not just checked before
                // the call. `biased` so a blackout that lands in the same
                // scheduling turn as the tool's completion wins: the point of
                // R3 is that nothing the agent does may observe a page the
                // person has taken over, and a result computed from that page
                // is exactly such an observation. Dropping the future cancels
                // the tool, and its result is discarded either way.
                let mut changes = self.subscribe_changes();
                let outcome = tokio::select! {
                    biased;
                    () = self.blackout_starts(&mut changes) => None,
                    out = exec.execute(tool, params) => Some(out),
                };
                // Paired with the `AgentActionStarted` above, on every exit.
                // Nothing used to clear it, so the drawer's action strip
                // reported the last browse action as still running for the
                // rest of the turn.
                self.apply_control(ControlEvent::AgentActionFinished).await;
                match outcome {
                    Some(out) => out,
                    None => Err(BLACKOUT_HOLDS_THE_PAGE.to_string()),
                }
            }
            _ => Err(format!("unknown tool: {tool}")),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::atomic::AtomicUsize;

    use crate::browser_attention::{
        RecordingAttention, BROWSER_SIGNIN_NEEDED, BROWSER_SIGNIN_RESOLVED,
    };

    /// A `BrowserTools` wired to a recorder under `conv-1`, exactly as
    /// `mcp_assistant::start` wires the real one under the run's key.
    fn tools_watching_signin() -> (BrowserTools, Arc<RecordingAttention>) {
        let tools = BrowserTools::new(std::env::temp_dir());
        let recorder = Arc::new(RecordingAttention::default());
        tools.set_signin_attention(recorder.clone(), Some("conv-1".to_string()));
        (tools, recorder)
    }

    /// The headline: an agent blocked at `browser_await_signin` reaches the
    /// operator through the always-on host channel, drawer or no drawer.
    #[tokio::test]
    async fn a_pending_sign_in_notifies_once_and_clears_once() {
        let (tools, recorder) = tools_watching_signin();
        tools.attach_agent_for_test().await;

        tools
            .apply_control_for_test(ControlEvent::SignInRequested(
                "Sign in at https://example.com/login".into(),
            ))
            .await;
        assert_eq!(
            recorder.calls(),
            vec![(
                BROWSER_SIGNIN_NEEDED.to_string(),
                Some("conv-1".to_string()),
                Some("Sign in at https://example.com/login".to_string()),
            )],
            "the conversation key and the tool's own prompt both travel"
        );

        tools
            .apply_control_for_test(ControlEvent::SignInResolved { signed_in: true })
            .await;
        assert_eq!(
            recorder.kinds(),
            vec![BROWSER_SIGNIN_NEEDED, BROWSER_SIGNIN_RESOLVED],
            "and the wait ending clears it"
        );
    }

    /// `presentation_pump` republishes and the resync sweep re-registers, so
    /// an emitter keyed on the apply rather than the TRANSITION would banner
    /// the operator every few seconds for one sign-in.
    #[tokio::test]
    async fn a_second_sign_in_request_while_one_is_pending_says_nothing() {
        let (tools, recorder) = tools_watching_signin();
        tools.attach_agent_for_test().await;

        for _ in 0..3 {
            tools
                .apply_control_for_test(ControlEvent::SignInRequested(
                    "Sign in at https://example.com/login".into(),
                ))
                .await;
        }
        assert_eq!(
            recorder.kinds(),
            vec![BROWSER_SIGNIN_NEEDED],
            "one wait is one notification"
        );

        // Nor does an unrelated transition (the user driving, the agent
        // labelling an action) produce one.
        tools.note_user_input().await;
        tools
            .apply_control_for_test(ControlEvent::AgentActionStarted("Filling the form".into()))
            .await;
        assert_eq!(recorder.kinds(), vec![BROWSER_SIGNIN_NEEDED]);
    }

    /// Every route that really clears a pending sign-in has to reach
    /// `resolved`. A badge that never clears is worse than no badge, and four
    /// of these five endings never touch the sign-in tool's own detection
    /// loop at all.
    #[tokio::test]
    async fn every_route_out_of_a_sign_in_reaches_resolved() {
        // `signed_in: true` — the detection loop saw the URL leave the login
        // flow. `signed_in: false` is also what the host-gone bail-out and a
        // failed navigate apply.
        for resolution in [
            ControlEvent::SignInResolved { signed_in: true },
            ControlEvent::SignInResolved { signed_in: false },
        ] {
            let (tools, recorder) = tools_watching_signin();
            tools.attach_agent_for_test().await;
            tools
                .apply_control_for_test(ControlEvent::SignInRequested("Sign in".into()))
                .await;
            tools.apply_control_for_test(resolution).await;
            assert_eq!(
                recorder.kinds(),
                vec![BROWSER_SIGNIN_NEEDED, BROWSER_SIGNIN_RESOLVED]
            );
        }

        // "Hand back to CAR" — the drawer's own affordance, which resolves a
        // pending sign-in as a side effect without the tool being told.
        let (tools, recorder) = tools_watching_signin();
        tools.attach_agent_for_test().await;
        tools
            .apply_control_for_test(ControlEvent::SignInRequested("Sign in".into()))
            .await;
        tools.hand_back().await;
        assert_eq!(
            recorder.kinds(),
            vec![BROWSER_SIGNIN_NEEDED, BROWSER_SIGNIN_RESOLVED],
            "hand-back is a real ending"
        );

        // The run ending under a sign-in nobody engaged with.
        let (tools, recorder) = tools_watching_signin();
        tools.attach_agent_for_test().await;
        tools
            .apply_control_for_test(ControlEvent::SignInRequested("Sign in".into()))
            .await;
        tools.note_run_ended().await;
        assert_eq!(
            recorder.kinds(),
            vec![BROWSER_SIGNIN_NEEDED, BROWSER_SIGNIN_RESOLVED],
            "a run ending must not leave a badge behind"
        );

        // The person engaged and then vanished: the run end DEFERS, and the
        // disconnect grace expiring is what finally settles it. Exactly one
        // resolved, at the right moment.
        let (tools, recorder) = tools_watching_signin();
        tools.attach_agent_for_test().await;
        tools
            .apply_control_for_test(ControlEvent::SignInRequested("Sign in".into()))
            .await;
        tools.note_user_input().await;
        tools.note_run_ended().await;
        assert_eq!(
            recorder.kinds(),
            vec![BROWSER_SIGNIN_NEEDED],
            "a person at the credential form still has a live window"
        );
        tools
            .apply_control_for_test(ControlEvent::GracePeriodExpired)
            .await;
        assert_eq!(
            recorder.kinds(),
            vec![BROWSER_SIGNIN_NEEDED, BROWSER_SIGNIN_RESOLVED]
        );
    }

    /// The `None` case: a `BrowserTools` with nowhere to report — `car do`,
    /// an embedder with no daemon, every other test in this file — behaves
    /// exactly as it did before, rather than panicking or blocking.
    #[tokio::test]
    async fn a_browser_with_no_attention_installed_still_works() {
        let tools = BrowserTools::new(std::env::temp_dir());
        tools.attach_agent_for_test().await;
        tools
            .apply_control_for_test(ControlEvent::SignInRequested("Sign in".into()))
            .await;
        assert!(tools.control_status().await.signin_pending);
        assert!(tools.control_status().await.blackout_active);
        tools.hand_back().await;
        assert!(!tools.control_status().await.signin_pending);
    }

    #[test]
    fn every_browser_tool_is_full_access() {
        // Browsing carries network egress and acts on a logged-in session, so
        // none of these may quietly land in a lower tier.
        for def in browser_tool_defs() {
            assert_eq!(
                def["tier"], BROWSER_TOOL_TIER,
                "{} must be full_access",
                def["name"]
            );
        }
    }

    #[test]
    fn record_tools_are_advertised_alongside_the_browse_tools() {
        let names: Vec<String> = browser_tool_defs()
            .iter()
            .filter_map(|d| d["name"].as_str().map(str::to_string))
            .collect();
        assert!(names.iter().any(|n| n == "browse_navigate"));
        assert!(names.iter().any(|n| n == "browser_record_start"));
        assert!(names.iter().any(|n| n == "browser_record_stop"));
    }

    #[tokio::test]
    async fn record_stop_without_start_is_an_error_not_a_panic() {
        let tools = BrowserTools::new(std::env::temp_dir());
        let err = tools
            .execute("browser_record_stop", &json!({}))
            .await
            .unwrap_err();
        assert!(err.contains("no recording in progress"), "got: {err}");
    }

    /// The regression this covers: `handle.stop()` succeeding used to leave
    /// the recording's Model consumer registered in `self.frames` — the CDP
    /// screencast stayed armed on a still page with no drawer open, until
    /// (if ever) some later page change happened to prune it. No live
    /// Chromium is needed: `frames.subscribe` only needs a browser BOUND to
    /// start CDP capture, not to register a consumer, and the recording is
    /// fed one real frame through its own channel rather than live capture.
    #[tokio::test]
    async fn record_stop_releases_its_frame_consumer() {
        // The encode step this test exercises for real shells out to ffmpeg
        // (see `run_record_stop_with`) — skip cleanly where it isn't
        // installed, rather than failing on tooling unrelated to this
        // regression.
        if std::process::Command::new("ffmpeg")
            .arg("-version")
            .output()
            .is_err()
        {
            eprintln!("skipping record_stop_releases_its_frame_consumer: ffmpeg not found");
            return;
        }

        let dir = tempfile::tempdir().expect("tempdir");
        let tools = BrowserTools::new(dir.path().to_path_buf());

        // Register a Model consumer on the same fan-out `run_record_start_with`
        // would, then simulate its owning task having ended — exactly what
        // `RecordingHandle::stop()`'s `self.task.abort()` does to the real
        // receiver in production. (The recording below is fed its frame
        // through a SEPARATE channel, since injecting a frame into this one
        // without live CDP capture isn't possible through the public API —
        // what matters for this test is that `self.frames` still shows this
        // consumer as registered-but-dead going into `run_record_stop_with`.)
        let (consumer_incoming, _epoch) = tools.frames.subscribe(FrameAudience::Model).await;
        assert_eq!(
            tools.frames.consumer_count_for_test().await,
            1,
            "precondition: the Model consumer is registered"
        );
        drop(consumer_incoming);

        // Build a real `RecordingHandle` with one real frame, so
        // `handle.stop()` inside `run_record_stop_with` succeeds instead of
        // erroring on an empty recording.
        let jpeg_path = dir.path().join("fixture.jpg");
        let status = std::process::Command::new("ffmpeg")
            .args([
                "-y",
                "-f",
                "lavfi",
                "-i",
                "color=c=white:s=2x2",
                "-frames:v",
                "1",
                "-q:v",
                "5",
            ])
            .arg(&jpeg_path)
            .status()
            .expect("run ffmpeg to build the fixture frame");
        assert!(status.success(), "ffmpeg must produce the fixture frame");
        let jpeg = std::fs::read(&jpeg_path).expect("read fixture frame");

        let (frame_tx, frame_rx) = tokio::sync::mpsc::channel(car_browser::FRAME_CHANNEL_CAP);
        frame_tx
            .try_send(car_browser::ScreencastFrame {
                jpeg: jpeg.into(),
                viewport: car_browser::Viewport {
                    width: 2,
                    height: 2,
                    device_pixel_ratio: 1.0,
                },
                captured_at: 0.0,
            })
            .expect("channel still open");
        drop(frame_tx);

        let handle = car_browser::recorder::start_with_frames(
            frame_rx,
            Instant::now(),
            &dir.path().join("recording"),
        )
        .await
        .expect("start_with_frames");
        *tools.recording.lock().await = Some(handle);
        // Give the recorder's background task a chance to actually drain
        // and write the one buffered frame before asking it to stop.
        tokio::time::sleep(Duration::from_millis(100)).await;

        let result = tools
            .run_record_stop_with(
                &json!({}),
                Duration::from_secs(5),
                Duration::from_millis(10),
            )
            .await;
        assert!(result.is_ok(), "record_stop should succeed: {result:?}");

        assert_eq!(
            tools.frames.consumer_count_for_test().await,
            0,
            "record_stop must prune the dead Model consumer, not leave the CDP \
             screencast armed indefinitely on a still page with no drawer open"
        );
    }

    /// The round-8 privacy blocker. A recording the agent never stopped kept
    /// writing after the run ended — and nothing tears the browser down at run
    /// end (deliberately), the person keeps driving the page (`browser_view`
    /// admits input once `owner == NoAgent`), and the recorder's Model consumer
    /// is deliberately kept registered so it RESUMES when a blackout lifts. So
    /// the frames still arriving were the person's own browsing, and they went
    /// to disk in the agent's recording directory.
    ///
    /// Run end stops it, and finalizes rather than discards: the model's later
    /// `browser_record_stop` still returns a video, one that ends where the run
    /// did.
    #[tokio::test]
    async fn a_run_ending_stops_the_recording_it_started() {
        if std::process::Command::new("ffmpeg")
            .arg("-version")
            .output()
            .is_err()
        {
            eprintln!("skipping a_run_ending_stops_the_recording_it_started: ffmpeg not found");
            return;
        }

        let dir = tempfile::tempdir().expect("tempdir");
        let tools = BrowserTools::new(dir.path().to_path_buf());
        tools.attach_agent_for_test().await;

        // The recording's own Model consumer, registered exactly as
        // `run_record_start_with` does, plus the quality override it installs.
        let (consumer_incoming, _epoch) = tools.frames.subscribe(FrameAudience::Model).await;
        let live_quality = tools.frames.quality();
        tools.frames.set_quality(1).await;
        *tools.recording_restore_quality.lock().await = Some(live_quality);
        drop(consumer_incoming);

        let jpeg_path = dir.path().join("fixture.jpg");
        let status = std::process::Command::new("ffmpeg")
            .args([
                "-y",
                "-f",
                "lavfi",
                "-i",
                "color=c=white:s=2x2",
                "-frames:v",
                "1",
                "-q:v",
                "5",
            ])
            .arg(&jpeg_path)
            .status()
            .expect("run ffmpeg to build the fixture frame");
        assert!(status.success(), "ffmpeg must produce the fixture frame");
        let jpeg = std::fs::read(&jpeg_path).expect("read fixture frame");

        let (frame_tx, frame_rx) = tokio::sync::mpsc::channel(car_browser::FRAME_CHANNEL_CAP);
        frame_tx
            .try_send(car_browser::ScreencastFrame {
                jpeg: jpeg.into(),
                viewport: car_browser::Viewport {
                    width: 2,
                    height: 2,
                    device_pixel_ratio: 1.0,
                },
                captured_at: 0.0,
            })
            .expect("channel still open");
        let handle = car_browser::recorder::start_with_frames(
            frame_rx,
            Instant::now(),
            &dir.path().join("recording"),
        )
        .await
        .expect("start_with_frames");
        *tools.recording.lock().await = Some(handle);
        tokio::time::sleep(Duration::from_millis(100)).await;

        // The run ends. The agent never called browser_record_stop.
        tools.note_run_ended().await;

        assert!(
            tools.recording.lock().await.is_none(),
            "no recording may still be running once the run that started it ended"
        );
        assert_eq!(
            tools.frames.consumer_count_for_test().await,
            0,
            "run end owes the same consumer release record_stop does"
        );
        assert_eq!(
            tools.frames.quality(),
            live_quality,
            "and the drawer gets its picture quality back"
        );

        // The property this whole blocker is about: anything Chrome emits AFTER
        // run-end — the person's own browsing — reaches no disk.
        let frames_on_disk = |dir: &std::path::Path| {
            std::fs::read_dir(dir)
                .map(|entries| {
                    entries
                        .filter_map(Result::ok)
                        .filter(|e| e.file_name().to_string_lossy().starts_with("frame-"))
                        .count()
                })
                .unwrap_or(0)
        };
        let recording_dir = dir.path().join("recording");
        tokio::time::sleep(Duration::from_millis(50)).await;
        let before = frames_on_disk(&recording_dir);
        assert_eq!(before, 1, "the pre-run-end frame was captured");
        let _ = frame_tx
            .send(car_browser::ScreencastFrame {
                jpeg: vec![0xff, 0xd8].into(),
                viewport: car_browser::Viewport {
                    width: 2,
                    height: 2,
                    device_pixel_ratio: 1.0,
                },
                captured_at: 1.0,
            })
            .await;
        tokio::time::sleep(Duration::from_millis(150)).await;
        assert_eq!(
            frames_on_disk(&recording_dir),
            before,
            "a frame emitted after the run ended must not be written to disk"
        );

        // Finalized, not discarded: the encode is still owed and still works.
        let result = tools
            .run_record_stop_with(
                &json!({}),
                Duration::from_secs(5),
                Duration::from_millis(10),
            )
            .await
            .expect("the frames captured up to run-end are still encodable");
        assert_eq!(
            result["frames"], 1,
            "one frame, the one from before run-end"
        );

        // And it is consumed exactly once.
        let err = tools
            .run_record_stop_with(
                &json!({}),
                Duration::from_secs(5),
                Duration::from_millis(10),
            )
            .await
            .unwrap_err();
        assert!(err.contains("no recording in progress"), "got: {err}");
    }

    #[test]
    fn describe_browse_action_humanizes_the_tool_name_and_leading_param() {
        assert_eq!(
            describe_browse_action("browse_navigate", &json!({"url": "https://x.com"})),
            "Navigate https://x.com"
        );
        assert_eq!(
            describe_browse_action(
                "browse_type",
                &json!({"element_id": "el_3", "text": "hello"})
            ),
            "Type hello"
        );
        // No recognized param present — still humanizes the tool name alone.
        assert_eq!(
            describe_browse_action("browse_scroll", &json!({})),
            "Scroll"
        );
    }

    #[tokio::test]
    async fn presentation_before_any_browser_call_is_the_no_agent_zero_ceremony_state() {
        let tools = BrowserTools::new(std::env::temp_dir());
        let snap = tools.presentation().await;
        assert_eq!(
            snap.owner,
            crate::assistant::browser_control::ControlOwner::NoAgent
        );
        assert!(snap.tabs.is_empty());
        assert!(!snap.blackout_active);
    }

    #[tokio::test]
    async fn take_control_without_an_agent_is_a_noop() {
        // "no ceremony" — nothing to take, matching the reducer's own
        // no-op behavior (see browser_control's take_control_is_a_noop_...
        // test). Exercised here through BrowserTools's own pass-through,
        // with no live Chromium involved.
        let tools = BrowserTools::new(std::env::temp_dir());
        let (snap, effects) = tools.take_control().await;
        assert_eq!(
            snap.owner,
            crate::assistant::browser_control::ControlOwner::NoAgent
        );
        assert!(effects.is_empty());
    }

    #[tokio::test]
    async fn wait_for_agent_turn_returns_immediately_once_attached() {
        let tools = BrowserTools::new(std::env::temp_dir());
        tools
            .presentation
            .lock()
            .await
            .apply_control(ControlEvent::AgentAttached);
        // A large timeout that would fail the test if this actually blocked.
        tools
            .wait_for_agent_turn_with(Duration::from_secs(5), Duration::from_millis(10))
            .await
            .expect("agent owns control, should not block");
    }

    #[tokio::test]
    async fn wait_for_agent_turn_times_out_while_the_user_holds_control() {
        let tools = BrowserTools::new(std::env::temp_dir());
        {
            let mut p = tools.presentation.lock().await;
            p.apply_control(ControlEvent::AgentAttached);
            p.apply_control(ControlEvent::TakeControl);
        }
        let err = tools
            .wait_for_agent_turn_with(Duration::from_millis(50), Duration::from_millis(10))
            .await
            .unwrap_err();
        assert!(err.contains("currently driving"), "got: {err}");
    }

    #[tokio::test]
    async fn wait_for_agent_turn_unblocks_once_the_user_hands_back() {
        let tools = Arc::new(BrowserTools::new(std::env::temp_dir()));
        {
            let mut p = tools.presentation.lock().await;
            p.apply_control(ControlEvent::AgentAttached);
            p.apply_control(ControlEvent::TakeControl);
        }
        let waiter = {
            let tools = Arc::clone(&tools);
            tokio::spawn(async move {
                tools
                    .wait_for_agent_turn_with(Duration::from_secs(5), Duration::from_millis(10))
                    .await
            })
        };
        tokio::time::sleep(Duration::from_millis(30)).await;
        tools.hand_back().await;
        waiter
            .await
            .expect("task panicked")
            .expect("should unblock once handed back, not time out");
    }

    // ---- profile isolation (FAIL 3/4/7) --------------------------------

    /// The regression that broke `browser.run`: setting
    /// `CAR_BROWSER_PROFILE_DIR` is process-global, so the drawer launching
    /// its standing session silently repointed every OTHER browser in the
    /// daemon — `browser.run`'s per-connection browsers included — at the
    /// agent's persistent profile, after which they all died on Chromium's
    /// SingletonLock for the life of that daemon.
    ///
    /// A grep-level test on purpose: the property is "this crate contains no
    /// code that writes that variable", which no behavioural test can state
    /// as directly, and which a future edit could reintroduce anywhere.
    #[test]
    fn car_code_never_sets_the_process_global_profile_env_var() {
        let src = include_str!("browser_tools.rs");
        // The name appears in prose in this file (this test included), so
        // look for the mutation, not the mention.
        assert!(
            !src.contains("set_var(\"CAR_BROWSER_PROFILE_DIR"),
            "the profile directory must be passed as a launch OPTION; setting the env var \
             repoints every other browser in the process"
        );
        assert!(
            !src.contains("set_var(&\"CAR_BROWSER_PROFILE_DIR"),
            "same, via a reference"
        );
    }

    #[test]
    fn the_agent_and_the_standing_session_use_different_persistent_profiles() {
        // Chromium allows exactly ONE live instance per profile directory,
        // so two browsers that can be alive at the same time must not share
        // one. This is the whole fix for "the standing session and an agent
        // browser cannot coexist".
        let agent = BrowserProfile::Agent.dir();
        let standing = BrowserProfile::StandingSession.dir();
        match (agent, standing) {
            (Some(a), Some(b)) => {
                assert_ne!(a, b, "two live browsers must not share one profile dir");
                assert_eq!(a.file_name().unwrap(), "browser-profile");
                assert_eq!(b.file_name().unwrap(), "browser-profile-user");
                assert_eq!(a.parent(), b.parent(), "siblings under the CAR state root");
            }
            // No CAR_HOME and no home directory: both fall back to
            // car-browser's throwaway per-instance profile, which cannot
            // collide with anything.
            (None, None) => {}
            other => panic!("profile dirs must resolve together or not at all: {other:?}"),
        }
    }

    #[tokio::test]
    async fn constructors_pick_persistent_and_isolated_profiles() {
        let agent = BrowserTools::new(std::env::temp_dir());
        let standing = BrowserTools::standing_session(std::env::temp_dir());
        let coder = BrowserTools::isolated(std::env::temp_dir());
        let other_coder = BrowserTools::isolated(std::env::temp_dir());
        assert!(!Arc::ptr_eq(&coder.inner, &other_coder.inner));
        assert!(!Arc::ptr_eq(&coder.inner, &agent.inner));
        assert_eq!(other_coder.profile, BrowserProfile::Isolated);
        assert_eq!(coder.profile, BrowserProfile::Isolated);
        assert!(coder.profile.launch_dir().is_none());
        assert_eq!(agent.profile, BrowserProfile::Agent);
        assert_eq!(standing.profile, BrowserProfile::StandingSession);
    }

    // ---- the agent attaches on first AGENT use, whoever launched ---------

    /// The regression this exists for: a user navigating in the drawer before
    /// the agent's first browse call used to launch the browser, and the
    /// agent's own `session()` would then find one already there and return
    /// without ever attaching — `owner` stuck at `NoAgent` for the whole run,
    /// so no strip, `take_control` a no-op, and the user-control blackout
    /// unable to engage at all.
    #[tokio::test]
    async fn a_user_launch_does_not_stop_the_agent_attaching_later() {
        let tools = BrowserTools::new(std::env::temp_dir());
        // Stand in for the user's navigation having already launched the
        // browser: ownership is what `ensure_session(false)` leaves alone.
        assert_eq!(
            tools.control_status().await.owner,
            crate::assistant::browser_control::ControlOwner::NoAgent
        );

        // The agent's first browse call, on a browser it did not launch.
        tools.attach_agent_for_test().await;
        assert_eq!(
            tools.control_status().await.owner,
            crate::assistant::browser_control::ControlOwner::Agent,
            "the agent must still attach on ITS first use"
        );
    }

    /// The other half of the same decision: the attach is once per run, not
    /// once per call. Firing it on every `session()` would hand control
    /// straight back to the agent the instant the user pressed Take control.
    #[tokio::test]
    async fn attaching_again_after_take_control_does_not_steal_control_back() {
        let tools = BrowserTools::new(std::env::temp_dir());
        tools.attach_agent_for_test().await;
        tools.take_control().await;
        assert_eq!(
            tools.control_status().await.owner,
            crate::assistant::browser_control::ControlOwner::User
        );

        // A second (and third) agent tool call in the same run.
        tools.attach_agent_for_test().await;
        tools.attach_agent_for_test().await;
        assert_eq!(
            tools.control_status().await.owner,
            crate::assistant::browser_control::ControlOwner::User,
            "the user keeps control until they hand it back"
        );
        assert!(tools.control_status().await.blackout_active);
    }

    /// The blocker this file's attach rule was rewritten for. ONE
    /// `BrowserTools` serves every turn of a `car do --serve` process, and
    /// `TurnGuard` fires `RunEnded` at the end of each one. With a latching
    /// once-per-instance flag, turn 2's browse call found `owner == NoAgent`,
    /// never re-attached, and sat in `wait_for_agent_turn` for the full
    /// 30-minute bound before returning a "the user is currently driving"
    /// error with no user involved.
    #[tokio::test]
    async fn the_next_turn_reattaches_after_the_previous_run_ended() {
        use crate::assistant::browser_control::ControlOwner;
        let tools = BrowserTools::new(std::env::temp_dir());

        // Turn 1.
        tools.attach_agent_for_test().await;
        assert_eq!(tools.control_status().await.owner, ControlOwner::Agent);
        tools.note_run_ended().await;
        assert_eq!(
            tools.control_status().await.owner,
            ControlOwner::NoAgent,
            "run end hands the browser back to the user, per the locked design"
        );

        // Turn 2, on the SAME instance. A timeout large enough that a
        // non-attaching implementation fails here instead of passing slowly.
        tools
            .wait_for_agent_turn_with(Duration::from_secs(5), Duration::from_millis(10))
            .await
            .expect("the next turn must re-attach, not wait out the pause");
        assert_eq!(tools.control_status().await.owner, ControlOwner::Agent);
    }

    /// The same boundary with a person holding the wheel across it: the run
    /// ends while the user has control (deferred by the reducer), so turn 2
    /// must NOT snatch it back — it waits, and hand-back is what lets it
    /// attach. Hand-back lands on `NoAgent` here (the previous run had
    /// ended), which is precisely the state the attach rule has to cover.
    #[tokio::test]
    async fn a_new_turn_waits_for_hand_back_instead_of_taking_the_wheel() {
        use crate::assistant::browser_control::ControlOwner;
        let tools = Arc::new(BrowserTools::new(std::env::temp_dir()));
        tools.attach_agent_for_test().await;
        tools.take_control().await;
        tools.note_run_ended().await;
        assert_eq!(
            tools.control_status().await.owner,
            ControlOwner::User,
            "a run ending under a person who is driving does not retract their control"
        );

        let waiter = {
            let tools = Arc::clone(&tools);
            tokio::spawn(async move {
                tools
                    .wait_for_agent_turn_with(Duration::from_secs(5), Duration::from_millis(10))
                    .await
            })
        };
        tokio::time::sleep(Duration::from_millis(30)).await;
        assert_eq!(
            tools.control_status().await.owner,
            ControlOwner::User,
            "the new turn must not have attached over the user"
        );
        tools.hand_back().await;
        waiter
            .await
            .expect("task panicked")
            .expect("hand-back must release the new turn, not leave it waiting");
        assert_eq!(tools.control_status().await.owner, ControlOwner::Agent);
    }

    // ---- blackout enforcement, model-facing half (R3) --------------------

    /// The mid-flight half of R3. `wait_while_blackout_with` answers only
    /// "may this call start", and `browse_wait` runs for as long as the model
    /// asks (`timeout_ms` is unclamped, and car-browser polls the whole page
    /// accessibility tree every 100ms until the deadline) — so a Take control
    /// ten seconds in would otherwise have the page the person just took over
    /// read back to the model when the call finally returned.
    ///
    /// This is the arm the browse call races its tool against.
    #[tokio::test]
    async fn a_blackout_starting_mid_call_resolves_the_cancellation_arm() {
        let tools = Arc::new(BrowserTools::new(std::env::temp_dir()));
        tools.attach_agent_for_test().await;
        let mut changes = tools.subscribe_changes();

        // No blackout: the arm must never resolve, or every browse call would
        // be cancelled the instant it started.
        assert!(
            tokio::time::timeout(
                Duration::from_millis(50),
                tools.blackout_starts(&mut changes)
            )
            .await
            .is_err(),
            "nothing is blacked out, so nothing may cancel"
        );

        // The user takes control while the tool is in flight.
        let presser = {
            let tools = Arc::clone(&tools);
            tokio::spawn(async move {
                tokio::time::sleep(Duration::from_millis(20)).await;
                tools.take_control().await;
            })
        };
        tokio::time::timeout(Duration::from_secs(5), tools.blackout_starts(&mut changes))
            .await
            .expect("a blackout starting mid-call must cancel the in-flight observation");
        presser.await.expect("task panicked");
        assert!(tools.control_status().await.blackout_active);
    }

    /// The action strip's label had no clear other than `RunEnded`, so it
    /// reported the last browse action as still running for the rest of the
    /// turn. The browse call now fires this on every exit — success, error,
    /// or blackout cancellation.
    #[tokio::test]
    async fn an_action_that_finished_stops_being_reported_as_current() {
        let tools = BrowserTools::new(std::env::temp_dir());
        tools.attach_agent_for_test().await;
        tools
            .apply_control(ControlEvent::AgentActionStarted("Opening x.test".into()))
            .await;
        assert_eq!(
            tools.presentation().await.current_action.as_deref(),
            Some("Opening x.test")
        );

        tools.apply_control(ControlEvent::AgentActionFinished).await;
        assert!(
            tools.presentation().await.current_action.is_none(),
            "a finished action must not keep claiming the strip"
        );
    }

    #[tokio::test]
    async fn a_pending_signin_suspends_model_facing_observation() {
        // The agent still nominally owns the browser here — `may_agent_act`
        // is true — so this gate is the ONLY thing standing between the
        // model and a page the user is typing a password into.
        let tools = BrowserTools::new(std::env::temp_dir());
        tools.apply_control(ControlEvent::AgentAttached).await;
        tools
            .apply_control(ControlEvent::SignInRequested("Sign in at x".into()))
            .await;
        assert!(
            tools.presentation.lock().await.control().may_agent_act(),
            "precondition: the agent owns control, so only the blackout gate applies"
        );

        let err = tools
            .wait_while_blackout_with(Duration::from_millis(50), Duration::from_millis(10))
            .await
            .unwrap_err();
        assert!(err.contains("cannot observe the page"), "got: {err}");
    }

    #[tokio::test]
    async fn observation_resumes_the_moment_the_signin_resolves() {
        let tools = Arc::new(BrowserTools::new(std::env::temp_dir()));
        tools.apply_control(ControlEvent::AgentAttached).await;
        tools
            .apply_control(ControlEvent::SignInRequested("Sign in at x".into()))
            .await;
        let waiter = {
            let tools = Arc::clone(&tools);
            tokio::spawn(async move {
                tools
                    .wait_while_blackout_with(Duration::from_secs(5), Duration::from_millis(10))
                    .await
            })
        };
        tokio::time::sleep(Duration::from_millis(30)).await;
        tools
            .apply_control(ControlEvent::SignInResolved { signed_in: true })
            .await;
        waiter
            .await
            .expect("task panicked")
            .expect("should unblock once the sign-in resolved, not time out");
    }

    #[tokio::test]
    async fn no_blackout_means_no_wait_at_all() {
        let tools = BrowserTools::new(std::env::temp_dir());
        tools.apply_control(ControlEvent::AgentAttached).await;
        // A timeout that would fail the test if this actually blocked.
        tools
            .wait_while_blackout_with(Duration::from_secs(5), Duration::from_millis(10))
            .await
            .expect("nothing is blacked out");
    }

    /// `browser_await_answer` measures the page and hands `content_length`
    /// back to the model, so it is model-facing observation and gates like
    /// the rest — at entry, before it can touch a session.
    #[tokio::test]
    async fn await_answer_is_blocked_at_entry_while_the_user_is_driving() {
        let tools = BrowserTools::new(std::env::temp_dir());
        tools.attach_agent_for_test().await;
        tools.take_control().await;

        let err = tools
            .run_await_answer_with(
                &json!({}),
                Duration::from_millis(50),
                Duration::from_millis(10),
            )
            .await
            .unwrap_err();
        // The gate's error, not a live-session one ("launch browser: …") —
        // proof it returned at the boundary and never reached session().
        assert!(err.contains("cannot observe the page"), "got: {err}");
        assert!(tools.inner.lock().await.is_none());
    }

    #[tokio::test]
    async fn await_answer_is_blocked_at_entry_while_a_signin_is_pending() {
        let tools = BrowserTools::new(std::env::temp_dir());
        tools.attach_agent_for_test().await;
        tools
            .apply_control_for_test(ControlEvent::SignInRequested("Sign in at x".into()))
            .await;

        let err = tools
            .run_await_answer_with(
                &json!({}),
                Duration::from_millis(50),
                Duration::from_millis(10),
            )
            .await
            .unwrap_err();
        assert!(err.contains("cannot observe the page"), "got: {err}");
    }

    /// The half an entry gate cannot cover: the user takes control (or a
    /// sign-in appears) while this is ALREADY polling. The loop must stop
    /// reading the page — this asserts on the measurement count itself, so it
    /// fails if the check is ever removed from the loop body.
    #[tokio::test]
    async fn await_answer_stops_measuring_the_page_when_a_blackout_starts_mid_poll() {
        let tools = Arc::new(BrowserTools::new(std::env::temp_dir()));
        tools.attach_agent_for_test().await;
        let reads = Arc::new(AtomicUsize::new(0));

        let settle = {
            let tools = Arc::clone(&tools);
            let reads = Arc::clone(&reads);
            tokio::spawn(async move {
                tools
                    .await_settled(
                        Duration::from_millis(400),
                        Duration::from_millis(10),
                        Duration::from_secs(60),
                        move || {
                            let reads = Arc::clone(&reads);
                            async move {
                                reads.fetch_add(1, Ordering::SeqCst);
                                // Keep growing, so nothing else could end the
                                // loop early — only the blackout can.
                                reads.load(Ordering::SeqCst) as i64 * 100
                            }
                        },
                    )
                    .await
            })
        };

        tokio::time::sleep(Duration::from_millis(80)).await;
        assert!(
            reads.load(Ordering::SeqCst) > 1,
            "precondition: it was actively measuring before the blackout"
        );
        tools.take_control().await;
        let at_blackout = reads.load(Ordering::SeqCst);
        tokio::time::sleep(Duration::from_millis(150)).await;
        assert_eq!(
            reads.load(Ordering::SeqCst),
            at_blackout,
            "not one page read may happen while the user is driving"
        );

        // And it resumes on hand-back rather than dying.
        tools.hand_back().await;
        tokio::time::sleep(Duration::from_millis(80)).await;
        assert!(
            reads.load(Ordering::SeqCst) > at_blackout,
            "measurement resumes once the blackout lifts"
        );
        let out = settle
            .await
            .expect("task panicked")
            .expect("returns a value");
        assert_eq!(out["settled"], false, "it ran out its own timeout");
    }

    /// The link that carries the reducer's blackout state to the frame
    /// fan-out — i.e. to `browser_record`'s disk output. Without it R3's
    /// recording half is unenforced no matter how correct the reducer is.
    #[tokio::test]
    async fn control_changes_republish_the_blackout_flag_to_the_frame_fanout() {
        let tools = BrowserTools::new(std::env::temp_dir());
        assert!(!tools.frames.blackout_active());

        tools.apply_control(ControlEvent::AgentAttached).await;
        assert!(
            !tools.frames.blackout_active(),
            "the agent driving is not a blackout"
        );

        tools.take_control().await;
        assert!(
            tools.frames.blackout_active(),
            "user control blacks out recording"
        );

        tools.hand_back().await;
        assert!(!tools.frames.blackout_active());

        tools
            .apply_control(ControlEvent::SignInRequested("Sign in".into()))
            .await;
        assert!(
            tools.frames.blackout_active(),
            "a pending sign-in blacks out too"
        );

        tools
            .apply_control(ControlEvent::SignInResolved { signed_in: true })
            .await;
        assert!(!tools.frames.blackout_active());
    }

    // ---- the drawer's change signal and user-input surface ---------------

    #[tokio::test]
    async fn every_control_change_wakes_the_drawer_signal() {
        let tools = BrowserTools::new(std::env::temp_dir());
        let mut changes = tools.subscribe_changes();
        assert!(!changes.has_changed().unwrap(), "nothing has happened yet");

        tools.apply_control(ControlEvent::AgentAttached).await;
        assert!(changes.has_changed().unwrap());
        changes.mark_unchanged();

        tools.take_control().await;
        assert!(changes.has_changed().unwrap());
    }

    #[tokio::test]
    async fn user_input_without_a_browser_is_a_clean_error_not_a_launch() {
        // Input that cannot sensibly launch a browser reports so, rather
        // than starting Chromium (or hanging) for a click with nothing to
        // click on.
        let tools = BrowserTools::new(std::env::temp_dir());
        for err in [
            tools.user_click(1.0, 2.0).await.unwrap_err(),
            tools.user_type("hi").await.unwrap_err(),
            tools.user_keypress("Enter", &[]).await.unwrap_err(),
            tools.user_scroll(10).await.unwrap_err(),
            tools.user_tab_close(fake_tab_id()).await.unwrap_err(),
            tools.user_tab_switch(fake_tab_id()).await.unwrap_err(),
            tools.resolve_tab("tab-0").await.unwrap_err(),
        ] {
            assert!(err.contains("no browser is running"), "got: {err}");
        }
        assert!(
            tools.inner.lock().await.is_none(),
            "none of those may have launched a browser"
        );
    }

    #[tokio::test]
    async fn navigate_rejects_an_empty_url_before_launching_anything() {
        let tools = BrowserTools::new(std::env::temp_dir());
        let err = tools.user_navigate("   ").await.unwrap_err();
        assert!(err.contains("non-empty `url`"), "got: {err}");
        assert!(tools.inner.lock().await.is_none());
    }

    /// A `TabId` the caller could plausibly have (the registry mints them),
    /// used only to reach the no-browser error path.
    fn fake_tab_id() -> car_browser::TabId {
        let (mut registry, _rx) = car_browser::tabs::TabRegistry::<&'static str>::new();
        registry.open("page", "about:blank", "")
    }

    // ---- wait_while_user_holds_control: the boundary gate for
    // browser_await_signin/browser_record_start/browser_record_stop ----

    #[tokio::test]
    async fn wait_while_user_holds_control_does_not_block_before_any_agent_attaches() {
        // The whole reason this is a DIFFERENT predicate from
        // may_agent_act: a call site gating before it has attached must
        // not mistake "nobody has attached yet" for "the user is driving."
        let tools = BrowserTools::new(std::env::temp_dir());
        tools
            .wait_while_user_holds_control_with(Duration::from_secs(5), Duration::from_millis(10))
            .await
            .expect("no agent attached yet, should not block");
    }

    #[tokio::test]
    async fn wait_while_user_holds_control_times_out_while_the_user_holds_control() {
        let tools = BrowserTools::new(std::env::temp_dir());
        {
            let mut p = tools.presentation.lock().await;
            p.apply_control(ControlEvent::AgentAttached);
            p.apply_control(ControlEvent::TakeControl);
        }
        let err = tools
            .wait_while_user_holds_control_with(
                Duration::from_millis(50),
                Duration::from_millis(10),
            )
            .await
            .unwrap_err();
        assert!(err.contains("currently driving"), "got: {err}");
    }

    #[tokio::test]
    async fn wait_while_user_holds_control_unblocks_once_the_user_hands_back() {
        let tools = Arc::new(BrowserTools::new(std::env::temp_dir()));
        {
            let mut p = tools.presentation.lock().await;
            p.apply_control(ControlEvent::AgentAttached);
            p.apply_control(ControlEvent::TakeControl);
        }
        let waiter = {
            let tools = Arc::clone(&tools);
            tokio::spawn(async move {
                tools
                    .wait_while_user_holds_control_with(
                        Duration::from_secs(5),
                        Duration::from_millis(10),
                    )
                    .await
            })
        };
        tokio::time::sleep(Duration::from_millis(30)).await;
        tools.hand_back().await;
        waiter
            .await
            .expect("task panicked")
            .expect("should unblock once handed back, not time out");
    }

    // ---- the actual call sites the reviewer flagged: browser_await_signin,
    // browser_record_start, browser_record_stop must consult the gate too,
    // not just browse_*. Each test drives the real method (via its `_with`
    // variant, so it doesn't have to wait AGENT_PAUSE_TIMEOUT_SECS) with the
    // user holding control and NO hand-back — proving the call site blocks
    // and then reports the boundary error, never reaching whatever it does
    // next (navigate / start a recording / touch a live session).

    #[tokio::test]
    async fn browser_await_signin_never_navigates_while_the_user_holds_control() {
        let tools = BrowserTools::new(std::env::temp_dir());
        {
            let mut p = tools.presentation.lock().await;
            p.apply_control(ControlEvent::AgentAttached);
            p.apply_control(ControlEvent::TakeControl);
        }
        let err = tools
            .run_await_signin_with(
                &json!({"url": "https://example.com/login"}),
                Duration::from_millis(50),
                Duration::from_millis(10),
            )
            .await
            .unwrap_err();
        // The gate's error, not a live-session error ("launch browser: ...")
        // — proof this returned at the boundary, before ever calling
        // session()/navigate. A real navigate attempt would also have taken
        // much longer than this test's ~50ms budget.
        assert!(err.contains("currently driving"), "got: {err}");
    }

    #[tokio::test]
    async fn browser_record_start_blocks_while_the_user_holds_control() {
        let tools = BrowserTools::new(std::env::temp_dir());
        {
            let mut p = tools.presentation.lock().await;
            p.apply_control(ControlEvent::AgentAttached);
            p.apply_control(ControlEvent::TakeControl);
        }
        let err = tools
            .run_record_start_with(
                &json!({}),
                Duration::from_millis(50),
                Duration::from_millis(10),
            )
            .await
            .unwrap_err();
        assert!(err.contains("currently driving"), "got: {err}");
    }

    #[tokio::test]
    async fn browser_record_stop_blocks_while_the_user_holds_control_and_then_re_checks() {
        // Full round trip, live-Chromium-free: no recording was ever
        // started, so once the gate releases (hand-back), this call
        // correctly falls through to the SAME "no recording in progress"
        // error the pre-existing (owner-agnostic) test asserts — proving
        // both that the gate blocks AND that the call site's own logic
        // still runs correctly afterward, not just that it errors out.
        let tools = Arc::new(BrowserTools::new(std::env::temp_dir()));
        {
            let mut p = tools.presentation.lock().await;
            p.apply_control(ControlEvent::AgentAttached);
            p.apply_control(ControlEvent::TakeControl);
        }
        let waiter = {
            let tools = Arc::clone(&tools);
            tokio::spawn(async move {
                tools
                    .run_record_stop_with(
                        &json!({}),
                        Duration::from_secs(5),
                        Duration::from_millis(10),
                    )
                    .await
            })
        };
        tokio::time::sleep(Duration::from_millis(30)).await;
        tools.hand_back().await;
        let err = waiter
            .await
            .expect("task panicked")
            .expect_err("no recording was ever started");
        assert!(err.contains("no recording in progress"), "got: {err}");
    }

    // ---- Task 7: the headless default flip + the sign-in-safe fallback --
    //
    // `decide_headless` is the whole rule as pure logic — no browser, no
    // daemon, no lock. Every other test below drives the real methods
    // (`effective_headless`, `run_await_signin_with`) but stops before
    // `session()` in every case, so none of these need a live Chromium
    // either.

    #[test]
    fn decide_headless_defaults_to_headless_when_a_host_is_connected() {
        // The whole point of the flip: the drawer is a real surface, so no
        // separate Chrome window.
        assert!(decide_headless(true, None));
    }

    #[test]
    fn decide_headless_defaults_to_headed_when_no_host_has_ever_connected() {
        // Pure CLI session — exactly today's behavior, unchanged.
        assert!(!decide_headless(false, None));
    }

    #[test]
    fn decide_headless_override_forces_headless_even_with_no_host() {
        // CAR_BROWSER_HEADLESS keeps its exact existing semantics: any
        // non-"0", non-empty value is truthy, and it wins over the
        // host-connected default in either direction.
        assert!(decide_headless(false, Some("1")));
        assert!(decide_headless(false, Some("yes")));
    }

    #[test]
    fn decide_headless_override_forces_headed_even_with_a_host_connected() {
        assert!(!decide_headless(true, Some("0")));
    }

    #[test]
    fn decide_headless_override_of_empty_string_is_falsy_like_unset() {
        // Matches the ORIGINAL `.map(|v| v != "0" && !v.is_empty())` reading
        // of the env var precisely — an empty value was never truthy.
        assert!(!decide_headless(true, Some("")));
        assert!(!decide_headless(false, Some("")));
    }

    // ---- signin_timeout_hint: naming a surface that actually exists ------

    #[test]
    fn signin_timeout_hint_names_the_window_when_headed() {
        // Headed never becomes headless mid-session, so the window text is
        // always accurate here regardless of host state.
        assert_eq!(
            signin_timeout_hint(false, true),
            "Ask the user to complete the login in the open browser window, then retry."
        );
        assert_eq!(
            signin_timeout_hint(false, false),
            "Ask the user to complete the login in the open browser window, then retry."
        );
    }

    #[test]
    fn signin_timeout_hint_names_the_drawer_when_headless_with_a_host() {
        assert_eq!(
            signin_timeout_hint(true, true),
            "Ask the user to complete the login in the CAR app's browser drawer, then retry."
        );
    }

    #[test]
    fn signin_timeout_hint_says_no_surface_when_headless_with_no_host() {
        // Neither a window (headless) nor a drawer (nobody connected to show
        // it in) exists — the old fixed text pointed the user at a window
        // that was never there in this case.
        let hint = signin_timeout_hint(true, false);
        assert!(hint.contains("open the CAR app"), "got: {hint}");
        assert!(
            !hint.contains("browser window"),
            "must not claim a window exists: {hint}"
        );
    }

    #[tokio::test]
    async fn effective_headless_before_any_launch_tracks_the_live_host_state() {
        let tools = BrowserTools::new(std::env::temp_dir());
        // No probe installed — "no way to know" reads as no host, matching
        // `any_host_connected`'s own default.
        assert!(!tools.effective_headless(false));
        assert!(tools.effective_headless(true));
    }

    #[tokio::test]
    async fn effective_headless_after_launch_is_fixed_regardless_of_current_host_state() {
        let tools = BrowserTools::new(std::env::temp_dir());
        // Simulate "this instance already launched headless" without a live
        // Chromium — the mid-session case the brief rules on explicitly:
        // launched headless, host since disconnected. The decision must NOT
        // flip back just because `host_connected` reads false now.
        let _ = tools.launched_headless.set(true);
        assert!(tools.effective_headless(false));
        assert!(tools.effective_headless(true));

        // And the other mid-session case: launched headed (no host at
        // launch), a host connects later. Stays headed for its lifetime.
        let headed = BrowserTools::new(std::env::temp_dir());
        let _ = headed.launched_headless.set(false);
        assert!(!headed.effective_headless(true));
    }

    #[tokio::test]
    async fn any_host_connected_reflects_whatever_probe_is_installed() {
        let tools = BrowserTools::new(std::env::temp_dir());
        assert!(!tools.host_connected_for_test().await, "no probe = no host");

        tools.set_host_connectivity(Arc::new(AlwaysConnected));
        assert!(tools.host_connected_for_test().await);
    }

    #[tokio::test]
    async fn set_host_connectivity_is_a_once_lock_the_first_install_wins() {
        let tools = BrowserTools::new(std::env::temp_dir());
        tools.set_host_connectivity(Arc::new(AlwaysConnected));
        let flag = Arc::new(AtomicBool::new(false));
        tools.set_host_connectivity(Arc::new(SharedHostConnected(Arc::clone(&flag))));
        // The second install is silently ignored — matches every production
        // call site, which installs exactly once before any browse call.
        assert!(tools.host_connected_for_test().await);
    }

    #[tokio::test]
    async fn shared_host_connected_reads_the_flag_live() {
        let flag = Arc::new(AtomicBool::new(false));
        let probe = SharedHostConnected(Arc::clone(&flag));
        assert!(!probe.any_host_connected().await);
        flag.store(true, Ordering::Release);
        assert!(probe.any_host_connected().await);
    }

    #[tokio::test]
    async fn browser_await_signin_reports_the_host_gone_result_when_headless_with_no_host() {
        // The seam's headline case: launched headless (a host was connected
        // at launch), no host connected now. No live Chromium involved —
        // this returns before `session()` is ever called.
        let tools = BrowserTools::new(std::env::temp_dir());
        let _ = tools.launched_headless.set(true);
        tools.attach_agent_for_test().await;

        let err = tools
            .run_await_signin_with(
                &json!({}),
                Duration::from_millis(50),
                Duration::from_millis(10),
            )
            .await
            .unwrap_err();
        assert_eq!(err, HOST_GONE_FOR_SIGNIN);
    }

    #[tokio::test]
    async fn browser_await_signin_proceeds_past_the_gate_when_a_host_is_connected() {
        // Same headless instance, but a host IS connected — Task 4's drawer
        // event handles the sign-in from here, so this must NOT return the
        // host-gone result. (It still can't reach a live Chromium in this
        // suite, so this only proves the gate itself lets it through — the
        // error it hits next is `session()`'s launch failure, not this one.)
        let tools = BrowserTools::new(std::env::temp_dir());
        let _ = tools.launched_headless.set(true);
        tools.set_host_connectivity(Arc::new(AlwaysConnected));
        tools.attach_agent_for_test().await;

        let host_connected = tools.any_host_connected().await;
        assert!(
            !tools.effective_headless(host_connected) || host_connected,
            "the gate must not fire while a host is connected"
        );
    }

    #[tokio::test]
    async fn browser_await_signin_never_reports_host_gone_for_a_headed_browser() {
        // Pure-CLI browser (launched headed, no host ever connected): the
        // window itself is the visible surface regardless of host state —
        // the host-gone message must never fire for it.
        let tools = BrowserTools::new(std::env::temp_dir());
        let _ = tools.launched_headless.set(false);
        tools.attach_agent_for_test().await;

        let host_connected = tools.any_host_connected().await;
        assert!(!tools.effective_headless(host_connected));
    }
}