LazyBoy2/crates/lazyboy-core/src/browser_client.rs

//! Session-owned asynchronous helper. Taking the process out of the slot makes cancellation safe:
//! dropping a request destroys that process group, and the next request starts fresh.
use anyhow::{anyhow, Context, Result};
use serde_json::{json, Value};
use std::{
    path::{Path, PathBuf},
    process::Stdio,
    time::Duration,
};
use tokio::{
    io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader},
    process::{Child, ChildStdin, ChildStdout},
};
#[derive(Debug, Default)]
pub struct BrowserClient {
    state: tokio::sync::Mutex<Option<Helper>>,
    timeout: Option<Duration>,
}
#[derive(Debug)]
struct Helper {
    child: Child,
    input: ChildStdin,
    output: BufReader<ChildStdout>,
    next: u64,
    container: Option<String>,
}
impl Drop for Helper {
    fn drop(&mut self) {
        #[cfg(unix)]
        if let Some(pid) = self.child.id() {
            // Playwright launches Chromium in a detached process group. Killing only
            // the Node helper group leaves those browser processes behind.
            let descendants = browser_descendants(pid);
            unsafe {
                libc::kill(-(pid as i32), libc::SIGKILL);
                for child in descendants.into_iter().rev() {
                    libc::kill(child as i32, libc::SIGKILL);
                }
            }
        }
        let _ = self.child.start_kill();
    }
}
#[cfg(unix)]
fn browser_descendants(root: u32) -> Vec<u32> {
    let Ok(output) = std::process::Command::new("ps")
        .args(["-axo", "pid=,ppid="])
        .output()
    else {
        return vec![];
    };
    let pairs = String::from_utf8_lossy(&output.stdout)
        .lines()
        .filter_map(|line| {
            let mut fields = line.split_whitespace();
            Some((
                fields.next()?.parse::<u32>().ok()?,
                fields.next()?.parse::<u32>().ok()?,
            ))
        })
        .collect::<Vec<_>>();
    let mut found = vec![root];
    let mut index = 0;
    while index < found.len() {
        let parent = found[index];
        for &(pid, ppid) in &pairs {
            if ppid == parent && !found.contains(&pid) {
                found.push(pid);
            }
        }
        index += 1;
    }
    found.remove(0);
    found
}
impl BrowserClient {
    pub async fn is_started(&self) -> bool {
        self.state.lock().await.is_some()
    }
    pub async fn close(&self) {
        let mut state = self.state.lock().await;
        if let Some(mut helper) = state.take() {
            // Graceful Chromium shutdown flushes the persistent profile.
            let _ = helper
                .input
                .write_all(b"{\"id\":\"close\",\"op\":\"close\"}\n")
                .await;
            let mut line = String::new();
            let _ =
                tokio::time::timeout(Duration::from_secs(3), helper.output.read_line(&mut line))
                    .await;
        }
    }
    pub async fn request_on(
        &self,
        cwd: &Path,
        profile: Option<PathBuf>,
        mut req: Value,
        container: Option<&str>,
    ) -> Result<Value> {
        let mut slot = self.state.lock().await;
        let mut helper = match slot.take() {
            Some(h) if container.is_none() || h.container.as_deref() == container => h,
            Some(_) | None => spawn(cwd, profile, container).await?,
        };
        helper.next += 1;
        let id = helper.next.to_string();
        req["id"] = json!(id);
        let reply = tokio::time::timeout(self.timeout.unwrap_or(Duration::from_secs(60)), async {
            helper
                .input
                .write_all(format!("{}\n", req).as_bytes())
                .await?;
            helper.input.flush().await?;
            let mut line = Vec::new();
            if (&mut helper.output)
                .take(4 * 1024 * 1024 + 1)
                .read_until(b'\n', &mut line)
                .await?
                == 0
            {
                let status = helper.child.wait().await?;
                if status.code() == Some(75) {
                    return Err(anyhow!(
                        "browser_busy: another task still owns this desktop browser; do independent work or retry after it releases the browser"
                    ));
                }
                return Err(anyhow!(
                    "browser helper exited; next request will restart it"
                ));
            }
            if line.len() > 4 * 1024 * 1024 {
                return Err(anyhow!("browser response too large"));
            }
            let value: Value = serde_json::from_slice(&line).context("browser response JSON")?;
            if value["id"].as_str() != Some(&id) {
                return Err(anyhow!("browser response id mismatch"));
            }
            Ok(value)
        })
        .await
        .map_err(|_| {
            anyhow!("browser helper timed out; its process was stopped; result may be unknown")
        })??;
        *slot = Some(helper);
        Ok(reply)
    }
}
async fn spawn(cwd: &Path, profile: Option<PathBuf>, seat_id: Option<&str>) -> Result<Helper> {
    if local_browser_enabled() {
        let script = crate::browser::find_helper_script(cwd)
            .ok_or_else(|| anyhow!("{}", crate::browser::INSTALL_HINT))?;
        return spawn_script(cwd, &script, profile).await;
    }
    let hub = crate::box_runtime::BoxPool::global().hub(seat_id.unwrap_or(crate::SESSION_SEAT));
    hub.ensure_browser_ready().await?;
    let mut command = tokio::process::Command::new("docker");
    command.args(["exec", "-i", "-w", "/workspace", "-e", "LAZYBOY_BROWSER_CDP=http://127.0.0.1:9222",
        hub.container(), "flock", "-E", "75", "-w", "5", "/home/box/.lazyboy-browser.lock", "node", "/opt/lazyboy/playwright/browser_helper.mjs"]);
    let mut helper = start_helper(command).await?;
    helper.container = Some(hub.seat_id().to_string());
    Ok(helper)
}

/// Local browser is an explicit compatibility mode, never a Docker fallback.
pub(crate) fn local_browser_enabled() -> bool {
    std::env::var("LAZYBOY_BROWSER_SURFACE").as_deref() == Ok("local")
}

async fn spawn_script(cwd: &Path, script: &Path, profile: Option<PathBuf>) -> Result<Helper> {
    let mut command = tokio::process::Command::new("node");
    command.arg(script).current_dir(cwd);
    if let Some(profile) = profile {
        command.env("LAZYBOY_BROWSER_PROFILE", profile);
    }
    start_helper(command).await
}
async fn start_helper(mut command: tokio::process::Command) -> Result<Helper> {
    command.stdin(Stdio::piped()).stdout(Stdio::piped()).stderr(Stdio::piped()).kill_on_drop(true);
    #[cfg(unix)]
    command.process_group(0);
    let mut child = command.spawn().context("start Playwright helper")?;
    let input = child.stdin.take().unwrap();
    let output = BufReader::new(child.stdout.take().unwrap());
    if let Some(mut err) = child.stderr.take() {
        tokio::spawn(async move {
            let _ = tokio::io::copy(&mut err, &mut tokio::io::sink()).await;
        });
    }
    Ok(Helper {
        child,
        input,
        output,
        next: 0,
        container: None,
    })
}

#[cfg(test)]
mod tests {
    use super::*;
    #[tokio::test]
    async fn dropping_helper_kills_detached_browser_descendants() {
        if std::process::Command::new("node").arg("--version").output().is_err() {
            eprintln!("SKIPPED: detached browser cleanup test requires Node.js on PATH");
            return;
        }
        let dir = std::env::temp_dir().join(format!("gb-detached-{}", uuid::Uuid::new_v4()));
        std::fs::create_dir_all(&dir).unwrap();
        let script = dir.join("fake.cjs");
        std::fs::write(&script, "const c=require('child_process').spawn(process.execPath,['-e','setInterval(()=>{},1000)'],{detached:true,stdio:'ignore'});require('fs').writeFileSync('child.pid',String(c.pid));setInterval(()=>{},1000)").unwrap();
        let helper = spawn_script(&dir, &script, None).await.unwrap();
        tokio::time::timeout(Duration::from_secs(5), async {
            while !dir.join("child.pid").exists() {
                tokio::time::sleep(Duration::from_millis(20)).await;
            }
        })
        .await
        .unwrap();
        let pid = std::fs::read_to_string(dir.join("child.pid"))
            .unwrap()
            .parse::<i32>()
            .unwrap();
        drop(helper);
        tokio::time::timeout(Duration::from_secs(5), async {
            while unsafe { libc::kill(pid, 0) } == 0 {
                tokio::time::sleep(Duration::from_millis(20)).await;
            }
        })
        .await
        .expect("detached browser child must exit when helper is dropped");
        std::fs::remove_dir_all(dir).unwrap();
    }
    #[tokio::test]
    async fn helper_timeout_mismatched_id_and_cancellation_allow_restart() {
        if std::process::Command::new("node")
            .arg("--version")
            .output()
            .is_err()
        {
            return;
        }
        let dir = std::env::temp_dir().join(format!("lazyboy-helper-{}", uuid::Uuid::new_v4()));
        std::fs::create_dir_all(&dir).unwrap();
        let script = dir.join("fake.cjs");
        std::fs::write(&script, "setInterval(()=>{},1000)").unwrap();
        let client = BrowserClient {
            state: tokio::sync::Mutex::new(Some(spawn_script(&dir, &script, None).await.unwrap())),
            timeout: Some(Duration::from_millis(250)),
        };
        let error = client
            .request_on(&dir, None, json!({"op":"ping"}), None)
            .await
            .unwrap_err();
        assert!(error.to_string().contains("timed out"));
        assert!(!client.is_started().await);
        std::fs::write(
            &script,
            "process.stdin.on('data',()=>console.log(JSON.stringify({id:'wrong',ok:true})))",
        )
        .unwrap();
        *client.state.lock().await = Some(spawn_script(&dir, &script, None).await.unwrap());
        assert!(client
            .request_on(&dir, None, json!({"op":"ping"}), None)
            .await
            .unwrap_err()
            .to_string()
            .contains("id mismatch"));
        assert!(!client.is_started().await);
        std::fs::write(&script, "setInterval(()=>{},1000)").unwrap();
        *client.state.lock().await = Some(spawn_script(&dir, &script, None).await.unwrap());
        assert!(tokio::time::timeout(
            Duration::from_millis(20),
            client.request_on(&dir, None, json!({"op":"ping"}), None)
        )
        .await
        .is_err());
        assert!(!client.is_started().await);
        // Restart the real local helper after failures; ping requires no Chromium.
        let real_script = crate::browser::find_helper_script(&dir).unwrap();
        *client.state.lock().await = Some(spawn_script(&dir, &real_script, None).await.unwrap());
        assert_eq!(
            client
                .request_on(&dir, None, json!({"op":"ping"}), None)
                .await
                .unwrap()["pong"],
            true
        );
        client.close().await;
        std::fs::remove_dir_all(dir).unwrap();
    }

    #[tokio::test]
    async fn helper_lock_timeout_is_reported_as_browser_busy() {
        if std::process::Command::new("node").arg("--version").output().is_err() {
            return;
        }
        let dir = std::env::temp_dir().join(format!("lazyboy-helper-busy-{}", uuid::Uuid::new_v4()));
        std::fs::create_dir_all(&dir).unwrap();
        let script = dir.join("busy.cjs");
        std::fs::write(&script, "process.stdin.once('data',()=>process.exit(75))").unwrap();
        let client = BrowserClient {
            state: tokio::sync::Mutex::new(Some(spawn_script(&dir, &script, None).await.unwrap())),
            timeout: Some(Duration::from_secs(2)),
        };
        let error = client
            .request_on(&dir, None, json!({"op":"ping"}), None)
            .await
            .unwrap_err();
        assert!(error.to_string().contains("browser_busy"), "{error:#}");
        std::fs::remove_dir_all(dir).unwrap();
    }
}