feat(worker): accept per-navigation readiness modes

crates/obscura-cli/src/worker.rs

@@ -1,7 +1,7 @@
 
 use std::sync::Arc;
 
-use obscura_browser::{BrowserContext, Page};
+use obscura_browser::{BrowserContext, Page, WaitUntil};
 use serde::{Deserialize, Serialize};
 use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
 
@@ -9,7 +9,11 @@ use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
 #[serde(tag = "cmd")]
 enum WorkerCommand {
     #[serde(rename = "navigate")]
-    Navigate { url: String },
+    Navigate {
+        url: String,
+        #[serde(default, rename = "waitUntil")]
+        wait_until: NavigationWaitUntil,
+    },
     #[serde(rename = "evaluate")]
     Evaluate { expression: String },
     #[serde(rename = "title")]
@@ -22,6 +26,31 @@ enum WorkerCommand {
     Shutdown,
 }
 
+/// Keep the wire protocol strict: WaitUntil::from_str falls back to Load.
+#[derive(Debug, Default, Deserialize)]
+enum NavigationWaitUntil {
+    #[default]
+    #[serde(rename = "load")]
+    Load,
+    #[serde(rename = "domcontentloaded")]
+    DomContentLoaded,
+    #[serde(rename = "networkidle0")]
+    NetworkIdle0,
+    #[serde(rename = "networkidle2")]
+    NetworkIdle2,
+}
+
+impl From<NavigationWaitUntil> for WaitUntil {
+    fn from(value: NavigationWaitUntil) -> Self {
+        match value {
+            NavigationWaitUntil::Load => Self::Load,
+            NavigationWaitUntil::DomContentLoaded => Self::DomContentLoaded,
+            NavigationWaitUntil::NetworkIdle0 => Self::NetworkIdle0,
+            NavigationWaitUntil::NetworkIdle2 => Self::NetworkIdle2,
+        }
+    }
+}
+
 #[derive(Debug, Serialize)]
 struct WorkerResponse {
     ok: bool,
@@ -96,8 +125,8 @@ async fn main() {
         };
 
         let resp = match cmd {
-            WorkerCommand::Navigate { url } => {
-                match page.navigate(&url).await {
+            WorkerCommand::Navigate { url, wait_until } => {
+                match page.navigate_with_wait(&url, wait_until.into()).await {
                     Ok(()) => WorkerResponse::success(serde_json::json!({
                         "title": page.title,
                         "url": page.url_string(),

crates/obscura-cli/tests/worker_navigation_wait.rs

@@ -0,0 +1,251 @@
+use std::io::{BufRead, BufReader, Read, Write};
+use std::net::{TcpListener, TcpStream};
+use std::process::{Child, ChildStdin, Command, Stdio};
+use std::sync::{
+    atomic::{AtomicBool, Ordering},
+    mpsc, Arc,
+};
+use std::thread::{self, JoinHandle};
+use std::time::Duration;
+
+use serde_json::{json, Value};
+
+struct Server {
+    address: String,
+    stop: Arc<AtomicBool>,
+    task: Option<JoinHandle<()>>,
+}
+
+impl Server {
+    fn new() -> Self {
+        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+        listener.set_nonblocking(true).unwrap();
+        let address = listener.local_addr().unwrap().to_string();
+        let stop = Arc::new(AtomicBool::new(false));
+        let stopped = stop.clone();
+        let task = thread::spawn(move || {
+            while !stopped.load(Ordering::Relaxed) {
+                match listener.accept() {
+                    Ok((stream, _)) => serve(stream),
+                    Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => {
+                        thread::sleep(Duration::from_millis(2));
+                    }
+                    Err(error) => panic!("accept fixture: {error}"),
+                }
+            }
+        });
+        Self {
+            address,
+            stop,
+            task: Some(task),
+        }
+    }
+
+    fn url(&self, path: &str) -> String {
+        format!("http://{}{path}", self.address)
+    }
+}
+
+impl Drop for Server {
+    fn drop(&mut self) {
+        self.stop.store(true, Ordering::Relaxed);
+        if let Some(task) = self.task.take() {
+            task.join().unwrap();
+        }
+    }
+}
+
+fn serve(mut stream: TcpStream) {
+    stream.set_nonblocking(false).unwrap();
+    stream
+        .set_read_timeout(Some(Duration::from_secs(2)))
+        .unwrap();
+    stream
+        .set_write_timeout(Some(Duration::from_secs(2)))
+        .unwrap();
+    let mut request = Vec::new();
+    let mut buffer = [0; 1024];
+    while !request.windows(4).any(|bytes| bytes == b"\r\n\r\n") {
+        let count = match stream.read(&mut buffer) {
+            Ok(count) => count,
+            Err(_) => return,
+        };
+        if count == 0 {
+            return;
+        }
+        request.extend_from_slice(&buffer[..count]);
+        assert!(request.len() < 8192, "fixture request too large");
+    }
+    let request = String::from_utf8_lossy(&request);
+    let path = request
+        .lines()
+        .next()
+        .unwrap()
+        .split_whitespace()
+        .nth(1)
+        .unwrap();
+    if path == "/disconnect" {
+        return;
+    }
+    let (status, headers, body) = (200, "", "<title>Ready</title><script>globalThis.inlineRan = true; setTimeout(() => { globalThis.laterRan = true; }, 350);</script><p>Body</p>");
+    let response = format!(
+        "HTTP/1.1 {status} Fixture\r\nContent-Type: text/html\r\n{headers}Content-Length: {}\r\nConnection: close\r\n\r\n{body}", body.len()
+    );
+    // A navigation cancellation may close the connection before this write.
+    let _ = stream.write_all(response.as_bytes());
+}
+
+struct Worker {
+    child: Child,
+    input: ChildStdin,
+    responses: mpsc::Receiver<Value>,
+    reader: Option<JoinHandle<()>>,
+}
+
+impl Worker {
+    fn new() -> Self {
+        Self::with_stealth(false)
+    }
+
+    fn with_stealth(stealth: bool) -> Self {
+        let mut child = Command::new(env!("CARGO_BIN_EXE_obscura-worker"))
+            .env("OBSCURA_ALLOW_PRIVATE_NETWORK", "1")
+            .env("OBSCURA_STEALTH", if stealth { "1" } else { "0" })
+            .env("OBSCURA_OBEY_ROBOTS", "0")
+            .env("OBSCURA_NAV_TIMEOUT_MS", "5000")
+            .env_remove("OBSCURA_PROXY")
+            .stdin(Stdio::piped())
+            .stdout(Stdio::piped())
+            .stderr(Stdio::inherit())
+            .spawn()
+            .unwrap();
+        let input = child.stdin.take().unwrap();
+        let output = child.stdout.take().unwrap();
+        let (sender, responses) = mpsc::channel();
+        let reader = thread::spawn(move || {
+            for line in BufReader::new(output).lines() {
+                let response = serde_json::from_str(&line.unwrap()).unwrap();
+                if sender.send(response).is_err() {
+                    break;
+                }
+            }
+        });
+        Self {
+            child,
+            input,
+            responses,
+            reader: Some(reader),
+        }
+    }
+
+    fn send(&mut self, command: Value) -> Value {
+        writeln!(self.input, "{command}").unwrap();
+        self.input.flush().unwrap();
+        self.responses
+            .recv_timeout(Duration::from_secs(10))
+            .expect("worker response before deadline")
+    }
+
+    fn navigate(&mut self, url: &str) -> Value {
+        let response = self.send(json!({ "cmd": "navigate", "url": url }));
+        assert_eq!(response["ok"], true, "{response}");
+        response["result"].clone()
+    }
+}
+
+impl Drop for Worker {
+    fn drop(&mut self) {
+        let _ = self.child.kill();
+        let _ = self.child.wait();
+        if let Some(reader) = self.reader.take() {
+            reader.join().unwrap();
+        }
+    }
+}
+
+fn check_modes(stealth: bool) {
+    let server = Server::new();
+    let mut worker = Worker::with_stealth(stealth);
+    for mode in ["load", "domcontentloaded", "networkidle0", "networkidle2"] {
+        let response =
+            worker.send(json!({"cmd": "navigate", "url": server.url("/"), "waitUntil": mode}));
+        assert_eq!(response["ok"], true, "{mode}: {response}");
+        assert_eq!(response["result"]["title"], "Ready");
+        let inline = worker.send(json!({"cmd": "evaluate", "expression": "globalThis.inlineRan"}));
+        assert_eq!(inline["result"], true, "parser script must run with {mode}");
+        if mode.starts_with("networkidle") {
+            let later = worker
+                .send(json!({"cmd": "evaluate", "expression": "globalThis.laterRan === true"}));
+            assert_eq!(
+                later["result"], true,
+                "{mode} must drive the event loop during its idle window"
+            );
+        }
+    }
+    assert_eq!(worker.navigate(&server.url("/"))["title"], "Ready");
+}
+
+#[test]
+fn every_readiness_mode_runs_parser_scripts_and_idle_modes_drive_timers() {
+    check_modes(false);
+}
+
+#[cfg(feature = "stealth")]
+#[test]
+fn readiness_modes_work_with_the_stealth_transport() {
+    check_modes(true);
+}
+
+#[test]
+fn omitted_mode_keeps_load_default_after_a_request_with_an_explicit_mode() {
+    let server = Server::new();
+    let mut worker = Worker::new();
+    worker.send(json!({"cmd": "navigate", "url": server.url("/"), "waitUntil": "networkidle0"}));
+    worker.navigate(&server.url("/"));
+    let later =
+        worker.send(json!({"cmd": "evaluate", "expression": "globalThis.laterRan === true"}));
+    assert_eq!(
+        later["result"], false,
+        "network-idle selection must not leak to the next request"
+    );
+}
+
+#[test]
+fn invalid_modes_are_errors_and_do_not_navigate_or_poison_the_worker() {
+    let mut worker = Worker::new();
+    worker.navigate("data:text/html,<title>Original</title>");
+    for mode in [
+        json!("domContentLoaded"),
+        json!("networkIdle"),
+        json!(""),
+        json!(null),
+        json!(1),
+        json!(true),
+    ] {
+        let response = worker.send(json!({"cmd": "navigate", "url": "data:text/html,<title>Changed</title>", "waitUntil": mode}));
+        assert_eq!(response["ok"], false, "{mode}: {response}");
+        assert!(response["error"]
+            .as_str()
+            .unwrap()
+            .starts_with("Invalid command:"));
+        assert_eq!(worker.send(json!({"cmd": "title"}))["result"], "Original");
+    }
+    assert_eq!(
+        worker.navigate("data:text/html,<title>Recovered</title>")["title"],
+        "Recovered"
+    );
+}
+
+#[test]
+fn readiness_modes_preserve_non_http_navigation() {
+    let mut worker = Worker::new();
+    for mode in ["load", "domcontentloaded", "networkidle0", "networkidle2"] {
+        for url in [
+            "about:blank",
+            "data:text/html,<title>Data</title><p>Local</p>",
+        ] {
+            let response = worker.send(json!({"cmd": "navigate", "url": url, "waitUntil": mode}));
+            assert_eq!(response["ok"], true, "{mode}: {response}");
+        }
+    }
+}

docs/CLI-reference.md

@@ -127,3 +127,25 @@ Default transport is stdio. See [Use the MCP server](Use-the-MCP-server.md).
 
 Render-enabled builds add `browser_screenshot` and `browser_pdf` to the MCP
 tool list. Streaming screencasts are available through CDP rather than MCP.
+
+## Worker navigation readiness
+
+`obscura-worker` accepts newline-delimited JSON commands on stdin and writes
+one response per command to stdout. Select a readiness mode per navigation:
+
+```json
+{"cmd":"navigate","url":"https://example.com","waitUntil":"domcontentloaded"}
+```
+
+`waitUntil` accepts `load`, `domcontentloaded`, `networkidle0`, or `networkidle2`.
+Omitting it preserves the existing `load` default. The option applies only to
+that request. Invalid values return an `ok: false` response without navigating;
+the worker remains available for subsequent commands.
+
+These modes use the existing `Page::navigate_with_wait` behavior. Parser scripts
+run in every mode. Network-idle modes additionally drive the event loop until
+active requests remain at or below zero or two, respectively, for 500 ms, with
+the engine's existing five-second idle deadline. `domcontentloaded` uses the
+engine's DOM readiness boundary; it does not change script execution or resource
+warmup semantics. Successful responses retain their existing `title` and `url`
+fields.