car-server-core 0.47.0

//! 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::PathBuf;
use std::sync::Arc;
use std::time::{Duration, Instant};

use async_trait::async_trait;
use car_browser::perception::vision::VisionPerceptionPipeline;
use car_browser::{BrowserBackend, BrowserToolExecutor, ChromiumBackend, RecordingHandle};
use car_engine::ToolExecutor;
use serde_json::{json, Value};
use tokio::sync::Mutex;

use crate::coder::policy::stays_under;

/// 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;

pub struct BrowserTools {
    root: PathBuf,
    /// Lazily launched. `None` until the first browse call.
    inner: Arc<Mutex<Option<Session>>>,
    recording: Arc<Mutex<Option<RecordingHandle>>>,
}

struct Session {
    backend: Arc<ChromiumBackend>,
    exec: BrowserToolExecutor,
}

impl BrowserTools {
    pub fn new(root: PathBuf) -> Self {
        Self {
            root,
            inner: Arc::new(Mutex::new(None)),
            recording: Arc::new(Mutex::new(None)),
        }
    }

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

    async fn session(&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.
            //
            // Only set when the caller hasn't chosen a profile, so an operator
            // who wants isolation (or parallel runs) can still opt out.
            if std::env::var_os("CAR_BROWSER_PROFILE_DIR").is_none() {
                if let Some(dir) = default_browser_profile_dir() {
                    let _ = std::fs::create_dir_all(&dir);
                    // SAFETY: set once, before any browser launch, on the
                    // assistant's own process.
                    unsafe { std::env::set_var("CAR_BROWSER_PROFILE_DIR", &dir) };
                }
            }
            // Headed, not headless: a recording is FOOTAGE — it should show the
            // app as a person sees it. Headless also trips bot-detection on
            // some login flows, which is exactly where a demo starts.
            let backend =
                ChromiumBackend::launch_with_options(car_browser::chromium::LaunchOptions {
                    width: VIEWPORT_W,
                    height: VIEWPORT_H,
                    headless: std::env::var("CAR_BROWSER_HEADLESS")
                        .map(|v| v != "0" && !v.is_empty())
                        .unwrap_or(false),
                    extra_args: Vec::new(),
                })
                .await
                .map_err(|e| format!("launch browser: {e}"))?;
            let backend = Arc::new(backend);
            *guard = Some(Session {
                backend: Arc::clone(&backend),
                exec: 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> {
        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);
        // 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 measure = || async {
            page.evaluate("document.body ? document.body.innerText.length : 0")
                .await
                .ok()
                .and_then(|v| v.into_value::<i64>().ok())
                .unwrap_or(0)
        };

        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() < Duration::from_secs(timeout) {
            tokio::time::sleep(poll).await;
            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.

    /// 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. Because the browser is HEADED,
    /// the user can just complete the flow — SSO, MFA, a device prompt,
    /// whatever it is — in the window that is already on screen. This tool is
    /// the handshake: 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> {
        let backend = self.session().await?;
        if let Some(url) = params.get("url").and_then(|v| v.as_str()) {
            if !url.trim().is_empty() {
                backend
                    .navigate(url)
                    .await
                    .map_err(|e| 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;
            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 {
                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.",
                }));
            }
        }
        Err(format!(
            "timed out after {timeout}s waiting for sign-in — the browser is still at {last}. \
             Ask the user to complete the login in the open browser window, then retry."
        ))
    }

    async fn run_record_start(&self, params: &Value) -> Result<Value, String> {
        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(),
            );
        }
        let backend = self.session().await?;
        let page = 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(80)
            .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)
        ));
        let handle = car_browser::recorder::start(&page, &dir, quality, 1, VIEWPORT_W, VIEWPORT_H)
            .await
            .map_err(|e| format!("start recording: {e}"))?;
        *rec = Some(handle);
        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> {
        let handle = self
            .recording
            .lock()
            .await
            .take()
            .ok_or("no recording in progress — call browser_record_start first")?;

        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}"))?;
        }

        let recording = handle
            .stop()
            .await
            .map_err(|e| format!("stop recording: {e}"))?;

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

/// `~/.car/browser-profile` — where a signed-in browser session lives between
/// runs. Returns None when no home directory resolves, in which case the
/// throwaway per-instance profile still applies.
fn default_browser_profile_dir() -> Option<PathBuf> {
    std::env::var_os("HOME")
        .or_else(|| std::env::var_os("USERPROFILE"))
        .map(|h| PathBuf::from(h).join(".car").join("browser-profile"))
}

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, in the browser window that is already on screen, \
            and wait until they have. Use this the moment a site needs authentication — you cannot \
            and must not type someone's credentials, but the browser is headed, so they can \
            complete any flow (SSO, MFA, a device prompt) themselves. TELL THE USER what to sign \
            into before calling this; 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, then delegate to car-browser's own
                // executor so the automation semantics live in one place.
                self.session().await?;
                let guard = self.inner.lock().await;
                let session = guard.as_ref().ok_or("browser session unavailable")?;
                session.exec.execute(tool, params).await
            }
            _ => Err(format!("unknown tool: {tool}")),
        }
    }
}

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

    #[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}");
    }
}