figmar/BrowserSkill
1
0
Fork 0
mirror of https://github.com/Tencent/BrowserSkill.git synced 2026-09-10 13:01:23 +08:00
Code Issues Packages Projects Releases Wiki Activity

merge: integrate main into scroll-to PR

Browse Source
This commit is contained in:
drakezhang
2026-09-09 23:47:27 +08:00
parent 7a10be21cb cff1ea62da
commit 29355e041d
342 changed files with 48377 additions and 6208 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/bsk-cli/tests/cli_parse.rs
+106
View File
@@ -6,6 +6,7 @@ use bsk::cli::daemon::{DaemonCmd, parse_duration};
use bsk::cli::navigate::NavigateCmd;
use bsk::cli::record::{RecordCmd, RecordSub};
use bsk::cli::session::{SessionCmd, SessionSub};
use bsk::cli::upload::UploadModeArg;
use bsk::{Cli, Command};
use clap::Parser;
@@ -13,6 +14,54 @@ fn parse(args: &[&str]) -> Cli {
Cli::try_parse_from(args).expect("clap parse should succeed")
}
#[test]
fn parses_upload_modes() {
let cli = parse(&[
"bsk",
"upload",
"@e1",
"--file",
"image.png",
"--session",
"s1",
]);
let Command::Upload(default_args) = cli.command else {
panic!("expected upload subcommand");
};
assert_eq!(default_args.mode, UploadModeArg::Input);
let cli = parse(&[
"bsk",
"upload",
"@e1",
"--file",
"image.png",
"--session",
"s1",
"--mode",
"drop",
]);
let Command::Upload(drop_args) = cli.command else {
panic!("expected upload subcommand");
};
assert_eq!(drop_args.mode, UploadModeArg::Drop);
assert!(
Cli::try_parse_from([
"bsk",
"upload",
"@e1",
"--file",
"image.png",
"--session",
"s1",
"--mode",
"auto",
])
.is_err()
);
}
#[test]
fn parses_daemon_start_with_defaults() {
let cli = parse(&["bsk", "daemon", "start"]);
@@ -215,6 +264,24 @@ fn parses_scroll_to_target() {
assert_eq!(args.target.as_deref(), Some("@e2"));
}
#[test]
fn parses_focus_target() {
let cli = parse(&["bsk", "focus", "@e2", "--session", "s1"]);
let Command::Focus(args) = cli.command else {
panic!("expected focus command");
};
assert_eq!(args.target.as_deref(), Some("@e2"));
}
#[test]
fn parses_blur_selector() {
let cli = parse(&["bsk", "blur", "--selector", "#search", "--session", "s1"]);
let Command::Blur(args) = cli.command else {
panic!("expected blur command");
};
assert_eq!(args.selector.as_deref(), Some("#search"));
}
#[test]
fn rejects_zero_click_count() {
assert!(
@@ -222,6 +289,45 @@ fn rejects_zero_click_count() {
);
}
#[test]
fn parses_upload_with_repeated_files() {
let cli = parse(&[
"bsk",
"upload",
"@e3",
"--file",
"one.png",
"--file",
"two.png",
"--session",
"s1",
]);
let Command::Upload(args) = cli.command else {
panic!("expected upload command");
};
assert_eq!(args.target.as_deref(), Some("@e3"));
assert_eq!(args.files.len(), 2);
}
#[test]
fn parses_download_with_exact_output_policy() {
let cli = parse(&[
"bsk",
"download",
"#export",
"--out",
"result.zip",
"--session",
"s1",
"--overwrite",
]);
let Command::Download(args) = cli.command else {
panic!("expected download command");
};
assert_eq!(args.target.as_deref(), Some("#export"));
assert!(args.overwrite);
}
#[test]
fn parses_record_start_with_browser_and_url() {
let cli = parse(&[
crates/bsk-cli/tests/handshake_compat.rs
+10 -10
View File
@@ -108,12 +108,12 @@ async fn send_handshake_with_floors(
async fn handshake_ok_when_protocol_matches() {
let (handle, _sock) = spawn_daemon().await;
let mut ws = open_ws(handle.ws_addr()).await;
let resp = send_handshake(&mut ws, "1.0", env!("CARGO_PKG_VERSION")).await;
let resp = send_handshake(&mut ws, "1.1", env!("CARGO_PKG_VERSION")).await;
let result: HandshakeResult = match resp.body {
ResponseBody::Ok(v) => serde_json::from_value(v).unwrap(),
ResponseBody::Err(e) => panic!("expected ok handshake, got {e:?}"),
};
assert_eq!(result.protocol_version, "1.0");
assert_eq!(result.protocol_version, "1.1");
assert_eq!(
result
.min_compatible_peer
@@ -135,7 +135,7 @@ async fn handshake_ok_when_app_versions_differ_but_protocol_matches() {
let (handle, _sock) = spawn_daemon().await;
let mut ws = open_ws(handle.ws_addr()).await;
let resp =
send_handshake_with_floors(&mut ws, "1.0", "9.9.9", Some("0.0.0"), Some("1.0")).await;
send_handshake_with_floors(&mut ws, "1.1", "9.9.9", Some("0.0.0"), Some("1.1")).await;
match resp.body {
ResponseBody::Ok(_) => {}
other => panic!("expected ok when protocol matches, got {other:?}"),
@@ -149,10 +149,10 @@ async fn handshake_skew_when_protocol_minor_differs() {
let mut ws = open_ws(handle.ws_addr()).await;
let resp = send_handshake_with_floors(
&mut ws,
"1.1",
"1.2",
env!("CARGO_PKG_VERSION"),
Some("0.0.0"),
Some("1.0"),
Some("1.1"),
)
.await;
match resp.body {
@@ -210,7 +210,7 @@ async fn handshake_legacy_ext_without_protocol_floor_still_ok() {
let mut ws = open_ws(handle.ws_addr()).await;
let resp = send_handshake_with_floors(
&mut ws,
"1.0",
"1.1",
env!("CARGO_PKG_VERSION"),
Some("0.1.0"),
None,
@@ -235,7 +235,7 @@ async fn status_surfaces_version_skew_for_skewed_browser() {
browser_name: "chrome".into(),
browser_version: "131.0".into(),
extension_version: "9.9.9".into(),
extension_protocol_version: "1.1".into(),
extension_protocol_version: "1.2".into(),
label: "Older".into(),
sink: bsk::daemon::browsers::BrowserSink { tx },
pending: Mutex::new(bsk::daemon::browsers::Pending::default()),
@@ -263,8 +263,8 @@ async fn status_surfaces_version_skew_for_skewed_browser() {
.iter()
.find(|s| s.instance_id == "skew-only-test")
.expect("status must list our skew client");
assert_eq!(skew.client_protocol_version, "1.1");
assert_eq!(skew.server_protocol_version, "1.0");
assert_eq!(skew.client_protocol_version, "1.2");
assert_eq!(skew.server_protocol_version, "1.1");
assert_eq!(skew.client_version, "9.9.9");
let entry = status
.browsers
@@ -281,7 +281,7 @@ async fn handshake_rejects_when_local_below_peer_min_compatible_protocol() {
let mut ws = open_ws(handle.ws_addr()).await;
let resp = send_handshake_with_floors(
&mut ws,
"1.0",
"1.1",
env!("CARGO_PKG_VERSION"),
Some("0.0.0"),
Some("99.0.0"),
crates/bsk-cli/tests/per_session_queue.rs
+2
View File
@@ -325,6 +325,7 @@ async fn session_stop_fast_fails_while_tool_is_in_flight() {
&state.session_interrupts,
&sid,
Duration::from_secs(5),
None,
)
.await;
assert!(
@@ -366,6 +367,7 @@ async fn session_stop_fast_fails_while_tool_is_in_flight() {
&state.session_interrupts,
&sid,
Duration::from_secs(5),
None,
)
.await
.expect("session.stop ok after tool completes");
crates/bsk-cli/tests/record_export_recovery.rs
+327
View File
@@ -0,0 +1,327 @@
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::Duration;
use bsk::daemon::info::DaemonInfo;
use bsk::daemon::paths::BSK_HOME_ENV;
use bsk::daemon::{self, DaemonConfig};
use bsk::ipc_client::IpcClient;
use bsk_protocol::system::{HandshakeParams, HandshakeResult};
use bsk_protocol::tools::{
RecordStopResult, RecordedTrace, RecorderInfo, SessionStartParams, SessionStartResult,
StopReason, TraceEntry, TraceStateV3, TraceV3, VOM_FORMAT_VERSION,
};
use bsk_protocol::{BrowserPeerInfo, Frame, Method, RequestFrame, ResponseBody, ResponseFrame};
use futures_util::{SinkExt, StreamExt};
use serde_json::json;
use tokio::process::Command;
use tokio_tungstenite::tungstenite::handshake::client::generate_key;
use tokio_tungstenite::tungstenite::http::Request;
use tokio_tungstenite::tungstenite::protocol::Message;
const TEST_EXT_ID: &str = "abcdefghijklmnopabcdefghijklmnop";
type Ws =
tokio_tungstenite::WebSocketStream<tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>>;
fn bsk_bin() -> PathBuf {
PathBuf::from(env!("CARGO_BIN_EXE_bsk"))
}
async fn connect_extension(addr: std::net::SocketAddr) -> Ws {
let origin = format!("chrome-extension://{TEST_EXT_ID}");
let request = Request::builder()
.method("GET")
.uri(format!("ws://{addr}/"))
.header("Host", addr.to_string())
.header("Upgrade", "websocket")
.header("Connection", "Upgrade")
.header("Sec-WebSocket-Version", "13")
.header("Sec-WebSocket-Key", generate_key())
.header("Origin", origin)
.body(())
.unwrap();
tokio_tungstenite::connect_async(request).await.unwrap().0
}
async fn handshake(ws: &mut Ws) {
let request = RequestFrame {
id: "handshake".into(),
method: Method::SystemHandshake,
params: Some(
serde_json::to_value(HandshakeParams {
client: "browser-skill-extension".into(),
version: "0.1.10".parse().unwrap(),
protocol_version: "1.1".into(),
instance_id: TEST_EXT_ID.into(),
browser: BrowserPeerInfo {
name: "chrome".into(),
version: "131.0".into(),
},
min_compatible_peer: Some("0.1.10".parse().unwrap()),
min_compatible_protocol: Some("1.1".into()),
label: "Record export recovery test".into(),
})
.unwrap(),
),
};
ws.send(Message::Text(serde_json::to_string(&request).unwrap()))
.await
.unwrap();
let Message::Text(response) = ws.next().await.unwrap().unwrap() else {
panic!("expected text handshake response");
};
let response: ResponseFrame = serde_json::from_str(&response).unwrap();
let ResponseBody::Ok(value) = response.body else {
panic!("handshake failed");
};
let _: HandshakeResult = serde_json::from_value(value).unwrap();
}
fn sample_trace() -> TraceV3 {
TraceV3 {
version: 3,
purpose: Some("export recovery".into()),
started_at: Some("2026-08-12T10:00:00Z".into()),
recorded_at: "2026-08-12T10:01:00Z".into(),
stopped_by: StopReason::CliStop,
entry: TraceEntry {
start_url: "https://example.com/".into(),
},
recorder: RecorderInfo {
bsk: "0.1.10".into(),
vom: VOM_FORMAT_VERSION,
},
states: vec![TraceStateV3 {
id: "s1".into(),
url: "https://example.com/".into(),
title: Some("Example".into()),
body: "@vom 1\nL1 page".into(),
truncated: false,
}],
steps: vec![],
}
}
fn run_extension(
mut ws: Ws,
record_stop_calls: Arc<AtomicUsize>,
session_stop_calls: Arc<AtomicUsize>,
) {
tokio::spawn(async move {
while let Some(Ok(Message::Text(text))) = ws.next().await {
let Ok(Frame::Request(request)) = serde_json::from_str::<Frame>(&text) else {
continue;
};
let body = match request.method {
Method::ToolSessionStart => {
let _: SessionStartParams =
serde_json::from_value(request.params.clone().unwrap()).unwrap();
ResponseBody::Ok(
serde_json::to_value(SessionStartResult {
agent_window_id: Some(100),
})
.unwrap(),
)
}
Method::ToolRecordStop => {
record_stop_calls.fetch_add(1, Ordering::SeqCst);
ResponseBody::Ok(
serde_json::to_value(RecordStopResult {
trace: RecordedTrace::V3(sample_trace()),
})
.unwrap(),
)
}
Method::ToolSessionStop => {
session_stop_calls.fetch_add(1, Ordering::SeqCst);
ResponseBody::Ok(json!({}))
}
_ => ResponseBody::Err(bsk_protocol::RpcError {
code: bsk_protocol::ErrorCode::ProtocolError,
message: format!("unexpected method {:?}", request.method),
data: None,
}),
};
let response = ResponseFrame {
id: request.id,
body,
};
ws.send(Message::Text(serde_json::to_string(&response).unwrap()))
.await
.unwrap();
}
});
}
async fn start_session(sock: &Path) -> String {
#[derive(serde::Deserialize)]
struct StartResult {
session_id: String,
}
let mut client = IpcClient::connect(sock).await.unwrap();
client
.call::<(), StartResult>(
"session-start",
Method::SessionStart,
None,
Duration::from_secs(5),
)
.await
.unwrap()
.unwrap()
.session_id
}
fn write_record_session(home: &Path, session_id: &str) {
std::fs::create_dir_all(home).unwrap();
std::fs::write(
home.join("record-session.json"),
format!("{{\"session_id\":\"{session_id}\",\"started_at\":\"1\"}}\n"),
)
.unwrap();
}
async fn run_record_stop(home: &Path, output: &Path) -> std::process::Output {
Command::new(bsk_bin())
.arg("record")
.arg("stop")
.arg("--output")
.arg(output)
.env(BSK_HOME_ENV, home)
.output()
.await
.unwrap()
}
async fn setup_daemon(temp: &Path) -> (bsk::daemon::DaemonHandle, PathBuf, PathBuf) {
let home = temp.join("bsk-home");
let sock = temp.join("daemon.sock");
let daemon = daemon::run(DaemonConfig::new(0), Some(sock.clone()))
.await
.unwrap();
bsk::daemon::info::write_to_path(
&DaemonInfo::now(
std::process::id(),
sock.clone(),
daemon.ws_addr().port(),
env!("CARGO_PKG_VERSION"),
),
&home.join("daemon.json"),
)
.unwrap();
(daemon, home, sock)
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn existing_trace_json_file_is_rejected_before_record_stop() {
let temp = tempfile::Builder::new()
.prefix("bsk-record-json-")
.tempdir()
.unwrap();
let (daemon, home, sock) = setup_daemon(temp.path()).await;
let output = temp.path().join("trace.json");
std::fs::write(&output, "{}\n").unwrap();
let mut extension = connect_extension(daemon.ws_addr()).await;
handshake(&mut extension).await;
let record_stop_calls = Arc::new(AtomicUsize::new(0));
let session_stop_calls = Arc::new(AtomicUsize::new(0));
run_extension(
extension,
Arc::clone(&record_stop_calls),
Arc::clone(&session_stop_calls),
);
let session_id = start_session(&sock).await;
write_record_session(&home, &session_id);
let first = run_record_stop(&home, &output).await;
let stderr = String::from_utf8_lossy(&first.stderr);
assert!(
!first.status.success(),
"legacy JSON --output should fail: {stderr}"
);
assert!(
stderr.contains("JSON file") || stderr.contains("bundle directory"),
"{stderr}"
);
let state: serde_json::Value =
serde_json::from_str(&std::fs::read_to_string(home.join("record-session.json")).unwrap())
.unwrap();
assert_eq!(state["session_id"], session_id);
assert_eq!(
record_stop_calls.load(Ordering::SeqCst),
0,
"existing trace.json must be rejected before the extension returns the Trace"
);
assert_eq!(session_stop_calls.load(Ordering::SeqCst), 0);
assert_eq!(std::fs::read_to_string(&output).unwrap(), "{}\n");
daemon.shutdown().await;
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn failed_bundle_export_can_be_recovered() {
let temp = tempfile::Builder::new()
.prefix("bsk-record-recov-")
.tempdir()
.unwrap();
let (daemon, home, sock) = setup_daemon(temp.path()).await;
let output = temp.path().join("trace");
std::fs::create_dir_all(&output).unwrap();
std::fs::write(output.join("states"), "not a directory\n").unwrap();
let mut extension = connect_extension(daemon.ws_addr()).await;
handshake(&mut extension).await;
let record_stop_calls = Arc::new(AtomicUsize::new(0));
let session_stop_calls = Arc::new(AtomicUsize::new(0));
run_extension(
extension,
Arc::clone(&record_stop_calls),
Arc::clone(&session_stop_calls),
);
let session_id = start_session(&sock).await;
write_record_session(&home, &session_id);
let first = run_record_stop(&home, &output).await;
let stderr = String::from_utf8_lossy(&first.stderr);
assert!(
!first.status.success(),
"blocked states path should fail export: {stderr}"
);
assert!(
home.join("record-recovery.json").exists(),
"completed Trace must be saved before the failed export"
);
assert!(
!home.join("record-session.json").exists(),
"session state is cleared after the extension returns the Trace"
);
assert_eq!(record_stop_calls.load(Ordering::SeqCst), 1);
assert_eq!(session_stop_calls.load(Ordering::SeqCst), 1);
std::fs::remove_file(output.join("states")).unwrap();
let second = run_record_stop(&home, &output).await;
assert!(
second.status.success(),
"recovery export failed: {}",
String::from_utf8_lossy(&second.stderr)
);
assert_eq!(
record_stop_calls.load(Ordering::SeqCst),
1,
"recovery must not call the extension again"
);
assert!(output.join("trace.json").exists());
assert!(output.join("states/s1.txt").exists());
assert!(
!home.join("record-recovery.json").exists(),
"successful export must delete recovery data"
);
daemon.shutdown().await;
}
crates/bsk-cli/tests/record_stop_retry.rs
+271
View File
@@ -0,0 +1,271 @@
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::Duration;
use bsk::cli::record_state;
use bsk::daemon::info::DaemonInfo;
use bsk::daemon::paths::BSK_HOME_ENV;
use bsk::daemon::{self, DaemonConfig};
use bsk::ipc_client::IpcClient;
use bsk_protocol::system::{HandshakeParams, HandshakeResult};
use bsk_protocol::tools::{
RecordStopResult, RecordedTrace, RecorderInfo, SessionStartParams, SessionStartResult,
StopReason, TraceEntry, TraceStateV3, TraceV3, VOM_FORMAT_VERSION,
};
use bsk_protocol::{
BrowserPeerInfo, ErrorCode, Frame, Method, RequestFrame, ResponseBody, ResponseFrame, RpcError,
};
use futures_util::{SinkExt, StreamExt};
use serde_json::json;
use tokio::process::Command;
use tokio_tungstenite::tungstenite::handshake::client::generate_key;
use tokio_tungstenite::tungstenite::http::Request;
use tokio_tungstenite::tungstenite::protocol::Message;
const TEST_EXT_ID: &str = "abcdefghijklmnopabcdefghijklmnop";
type Ws =
tokio_tungstenite::WebSocketStream<tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>>;
fn bsk_bin() -> PathBuf {
PathBuf::from(env!("CARGO_BIN_EXE_bsk"))
}
async fn connect_extension(addr: std::net::SocketAddr) -> Ws {
let origin = format!("chrome-extension://{TEST_EXT_ID}");
let request = Request::builder()
.method("GET")
.uri(format!("ws://{addr}/"))
.header("Host", addr.to_string())
.header("Upgrade", "websocket")
.header("Connection", "Upgrade")
.header("Sec-WebSocket-Version", "13")
.header("Sec-WebSocket-Key", generate_key())
.header("Origin", origin)
.body(())
.unwrap();
tokio_tungstenite::connect_async(request).await.unwrap().0
}
async fn handshake(ws: &mut Ws) {
let request = RequestFrame {
id: "handshake".into(),
method: Method::SystemHandshake,
params: Some(
serde_json::to_value(HandshakeParams {
client: "browser-skill-extension".into(),
version: "0.1.10".parse().unwrap(),
protocol_version: "1.1".into(),
instance_id: TEST_EXT_ID.into(),
browser: BrowserPeerInfo {
name: "chrome".into(),
version: "131.0".into(),
},
min_compatible_peer: Some("0.1.10".parse().unwrap()),
min_compatible_protocol: Some("1.1".into()),
label: "Record stop retry test".into(),
})
.unwrap(),
),
};
ws.send(Message::Text(serde_json::to_string(&request).unwrap()))
.await
.unwrap();
let Message::Text(response) = ws.next().await.unwrap().unwrap() else {
panic!("expected text handshake response");
};
let response: ResponseFrame = serde_json::from_str(&response).unwrap();
let ResponseBody::Ok(value) = response.body else {
panic!("handshake failed");
};
let _: HandshakeResult = serde_json::from_value(value).unwrap();
}
fn sample_trace() -> TraceV3 {
TraceV3 {
version: 3,
purpose: Some("retry record stop".into()),
started_at: Some("2026-08-12T10:00:00Z".into()),
recorded_at: "2026-08-12T10:01:00Z".into(),
stopped_by: StopReason::CliStop,
entry: TraceEntry {
start_url: "https://example.com/".into(),
},
recorder: RecorderInfo {
bsk: "0.1.10".into(),
vom: VOM_FORMAT_VERSION,
},
states: vec![TraceStateV3 {
id: "s1".into(),
url: "https://example.com/".into(),
title: Some("Example".into()),
body: "@vom 1\nL1 page".into(),
truncated: false,
}],
steps: vec![],
}
}
fn run_extension(
mut ws: Ws,
record_stop_calls: Arc<AtomicUsize>,
session_stop_calls: Arc<AtomicUsize>,
) {
tokio::spawn(async move {
while let Some(Ok(Message::Text(text))) = ws.next().await {
let Ok(Frame::Request(request)) = serde_json::from_str::<Frame>(&text) else {
continue;
};
let body = match request.method {
Method::ToolSessionStart => {
let _: SessionStartParams =
serde_json::from_value(request.params.clone().unwrap()).unwrap();
ResponseBody::Ok(
serde_json::to_value(SessionStartResult {
agent_window_id: Some(100),
})
.unwrap(),
)
}
Method::ToolRecordStop => {
if record_stop_calls.fetch_add(1, Ordering::SeqCst) == 0 {
ResponseBody::Err(RpcError {
code: ErrorCode::ProtocolError,
message: "recording remains active; retry record stop".into(),
data: Some(json!({ "recording": true, "retryable": true })),
})
} else {
ResponseBody::Ok(
serde_json::to_value(RecordStopResult {
trace: RecordedTrace::V3(sample_trace()),
})
.unwrap(),
)
}
}
Method::ToolSessionStop => {
session_stop_calls.fetch_add(1, Ordering::SeqCst);
ResponseBody::Ok(json!({}))
}
_ => ResponseBody::Err(RpcError {
code: ErrorCode::ProtocolError,
message: format!("unexpected method {:?}", request.method),
data: None,
}),
};
let response = ResponseFrame {
id: request.id,
body,
};
ws.send(Message::Text(serde_json::to_string(&response).unwrap()))
.await
.unwrap();
}
});
}
async fn start_session(sock: &Path) -> String {
#[derive(serde::Deserialize)]
struct StartResult {
session_id: String,
}
let mut client = IpcClient::connect(sock).await.unwrap();
client
.call::<(), StartResult>(
"session-start",
Method::SessionStart,
None,
Duration::from_secs(5),
)
.await
.unwrap()
.unwrap()
.session_id
}
async fn run_record_stop(home: &Path, output: &Path) -> std::process::Output {
Command::new(bsk_bin())
.arg("record")
.arg("stop")
.arg("--output")
.arg(output)
.env(BSK_HOME_ENV, home)
.output()
.await
.unwrap()
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn failed_record_stop_keeps_state_and_session_for_retry() {
// Keep the Unix-domain socket below macOS's SUN_LEN even from a deep worktree.
let temp = tempfile::Builder::new()
.prefix("bsk-record-stop-")
.tempdir()
.unwrap();
let home = temp.path().join("bsk-home");
let sock = temp.path().join("daemon.sock");
let output = temp.path().join("trace");
let daemon = daemon::run(DaemonConfig::new(0), Some(sock.clone()))
.await
.unwrap();
bsk::daemon::info::write_to_path(
&DaemonInfo::now(
std::process::id(),
sock.clone(),
daemon.ws_addr().port(),
env!("CARGO_PKG_VERSION"),
),
&home.join("daemon.json"),
)
.unwrap();
let mut extension = connect_extension(daemon.ws_addr()).await;
handshake(&mut extension).await;
let record_stop_calls = Arc::new(AtomicUsize::new(0));
let session_stop_calls = Arc::new(AtomicUsize::new(0));
run_extension(
extension,
Arc::clone(&record_stop_calls),
Arc::clone(&session_stop_calls),
);
let session_id = start_session(&sock).await;
unsafe {
std::env::set_var(BSK_HOME_ENV, &home);
}
record_state::write(&session_id).unwrap();
let first = run_record_stop(&home, &output).await;
assert!(
!first.status.success(),
"first stop unexpectedly succeeded: {}",
String::from_utf8_lossy(&first.stdout)
);
assert_eq!(record_state::read().unwrap().session_id, session_id);
assert_eq!(record_stop_calls.load(Ordering::SeqCst), 1);
assert_eq!(
session_stop_calls.load(Ordering::SeqCst),
0,
"failed record stop must leave the session active"
);
let second = run_record_stop(&home, &output).await;
assert!(
second.status.success(),
"retry failed: {}",
String::from_utf8_lossy(&second.stderr)
);
assert!(record_state::read().is_err());
assert_eq!(record_stop_calls.load(Ordering::SeqCst), 2);
assert_eq!(session_stop_calls.load(Ordering::SeqCst), 1);
assert!(output.join("trace.json").exists());
assert!(output.join("states/s1.txt").exists());
unsafe {
std::env::remove_var(BSK_HOME_ENV);
}
daemon.shutdown().await;
}
crates/bsk-cli/tests/sessions_ipc.rs
+368 -10
View File
@@ -10,9 +10,14 @@ use std::time::Duration;
use bsk::daemon::{self, DaemonConfig};
use bsk::ipc_client::IpcClient;
use bsk_protocol::cancel::{CancelParams, CancelResult};
use bsk_protocol::system::{HandshakeParams, HandshakeResult, StatusResult};
use bsk_protocol::tools::{SessionStartParams, SessionStartResult, SessionStopParams};
use bsk_protocol::{BrowserPeerInfo, Frame, Method, RequestFrame, ResponseBody, ResponseFrame};
use bsk_protocol::tools::{
SessionStartParams, SessionStartResult, SessionStopParams, SessionStopResult,
};
use bsk_protocol::{
BrowserPeerInfo, ErrorCode, Frame, Method, RequestFrame, ResponseBody, ResponseFrame, RpcError,
};
use futures_util::{SinkExt, StreamExt};
use rand::Rng;
use tokio_tungstenite::tungstenite::handshake::client::generate_key;
@@ -23,6 +28,9 @@ use support::{wait_for_browser_count, wait_for_no_sessions};
const TEST_EXT_ID: &str = "abcdefghijklmnopabcdefghijklmnop";
type TestWs =
tokio_tungstenite::WebSocketStream<tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>>;
fn tempfile_path(prefix: &str) -> PathBuf {
let mut p = std::env::temp_dir();
let mut rng = rand::thread_rng();
@@ -56,9 +64,7 @@ async fn spawn_daemon_with_session_idle(session_idle: Duration) -> (daemon::Daem
(handle, sock)
}
async fn connect_ext(
addr: std::net::SocketAddr,
) -> tokio_tungstenite::WebSocketStream<tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>> {
async fn connect_ext(addr: std::net::SocketAddr) -> TestWs {
let origin = format!("chrome-extension://{TEST_EXT_ID}");
let url = format!("ws://{addr}/");
let req = Request::builder()
@@ -76,11 +82,7 @@ async fn connect_ext(
ws
}
async fn handshake_as_ext(
ws: &mut tokio_tungstenite::WebSocketStream<
tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>,
>,
) -> HandshakeResult {
async fn handshake_as_ext(ws: &mut TestWs) -> HandshakeResult {
let params = HandshakeParams {
client: "browser-skill-extension".into(),
version: "0.1.0-dev.0".parse().unwrap(),
@@ -114,6 +116,128 @@ async fn handshake_as_ext(
}
}
async fn next_extension_request(ws: &mut TestWs) -> RequestFrame {
loop {
let message = ws.next().await.expect("daemon closed websocket").unwrap();
match message {
Message::Text(text) => match serde_json::from_str::<Frame>(&text).unwrap() {
Frame::Request(request) => return request,
Frame::Response(_) | Frame::Event(_) => continue,
},
Message::Ping(payload) => ws.send(Message::Pong(payload)).await.unwrap(),
Message::Close(_) => panic!("daemon closed websocket before sending request"),
_ => {}
}
}
}
async fn send_extension_response(ws: &mut TestWs, response: ResponseFrame) {
ws.send(Message::Text(serde_json::to_string(&response).unwrap()))
.await
.unwrap();
}
async fn acknowledge_extension_cancel(ws: &mut TestWs, cancel: RequestFrame, target_rpc_id: &str) {
assert_eq!(cancel.method, Method::Cancel);
let params: CancelParams = serde_json::from_value(cancel.params.unwrap()).unwrap();
assert_eq!(params.rpc_id, target_rpc_id);
send_extension_response(
ws,
ResponseFrame {
id: cancel.id,
body: ResponseBody::Ok(serde_json::to_value(CancelResult { cancelled: true }).unwrap()),
},
)
.await;
}
async fn respond_to_aborted_start(
ws: &mut TestWs,
start_seen: Option<tokio::sync::oneshot::Sender<()>>,
agent_window_id: i64,
) {
let start = next_extension_request(ws).await;
assert_eq!(start.method, Method::ToolSessionStart);
let params: SessionStartParams = serde_json::from_value(start.params.clone().unwrap()).unwrap();
if let Some(start_seen) = start_seen {
start_seen.send(()).unwrap();
}
let cancel = next_extension_request(ws).await;
acknowledge_extension_cancel(ws, cancel, &start.id).await;
// Model the hardest race: chrome.windows.create completed just as
// cancellation arrived, so the extension's original call succeeds.
send_extension_response(
ws,
ResponseFrame {
id: start.id,
body: ResponseBody::Ok(
serde_json::to_value(SessionStartResult {
agent_window_id: Some(agent_window_id),
})
.unwrap(),
),
},
)
.await;
let rollback = next_extension_request(ws).await;
assert_eq!(rollback.method, Method::ToolSessionStop);
let rollback_params: SessionStopParams =
serde_json::from_value(rollback.params.unwrap()).unwrap();
assert_eq!(rollback_params.session_id, params.session_id);
send_extension_response(
ws,
ResponseFrame {
id: rollback.id,
body: ResponseBody::Ok(serde_json::to_value(SessionStopResult::default()).unwrap()),
},
)
.await;
}
async fn respond_to_aborted_stop(
ws: &mut TestWs,
stop_seen: Option<tokio::sync::oneshot::Sender<()>>,
) {
let start = next_extension_request(ws).await;
assert_eq!(start.method, Method::ToolSessionStart);
send_extension_response(
ws,
ResponseFrame {
id: start.id,
body: ResponseBody::Ok(
serde_json::to_value(SessionStartResult {
agent_window_id: Some(4242),
})
.unwrap(),
),
},
)
.await;
let stop = next_extension_request(ws).await;
assert_eq!(stop.method, Method::ToolSessionStop);
if let Some(stop_seen) = stop_seen {
stop_seen.send(()).unwrap();
}
let cancel = next_extension_request(ws).await;
acknowledge_extension_cancel(ws, cancel, &stop.id).await;
send_extension_response(
ws,
ResponseFrame {
id: stop.id,
body: ResponseBody::Err(RpcError {
code: ErrorCode::Cancelled,
message: "session_stop aborted before window close".into(),
data: None,
}),
},
)
.await;
}
#[tokio::test]
async fn session_start_stop_round_trip_via_ipc() {
let (handle, sock) = spawn_daemon().await;
@@ -250,6 +374,240 @@ async fn session_start_stop_round_trip_via_ipc() {
handle.shutdown().await;
}
#[tokio::test]
async fn cancelling_session_start_rolls_back_a_late_extension_success() {
const START_RPC_ID: &str = "session-start-cancel";
let (handle, sock) = spawn_daemon().await;
let mut ws = connect_ext(handle.ws_addr()).await;
let _ = handshake_as_ext(&mut ws).await;
let (start_seen_tx, start_seen_rx) = tokio::sync::oneshot::channel();
let (release_tx, release_rx) = tokio::sync::oneshot::channel();
let responder = tokio::spawn(async move {
respond_to_aborted_start(&mut ws, Some(start_seen_tx), 4242).await;
let _ = release_rx.await;
});
let mut start_ipc = IpcClient::connect(&sock).await.unwrap();
let start_call = tokio::spawn(async move {
let outcome: Result<serde_json::Value, RpcError> = start_ipc
.call_with_id(
START_RPC_ID.to_string(),
Method::SessionStart,
Some(serde_json::json!({ "focused": false })),
Duration::from_secs(5),
)
.await
.unwrap();
outcome
});
start_seen_rx.await.unwrap();
let mut cancel_ipc = IpcClient::connect(&sock).await.unwrap();
let cancel: CancelResult = cancel_ipc
.call(
"cancel-start",
Method::Cancel,
Some(CancelParams {
rpc_id: START_RPC_ID.to_string(),
}),
Duration::from_secs(2),
)
.await
.unwrap()
.unwrap();
assert!(cancel.cancelled);
let start_error = start_call.await.unwrap().unwrap_err();
assert_eq!(start_error.code, ErrorCode::Cancelled);
let mut status_ipc = IpcClient::connect(&sock).await.unwrap();
let status: StatusResult = status_ipc
.call::<(), _>(
"status-after-start-cancel",
Method::SystemStatus,
None,
Duration::from_secs(2),
)
.await
.unwrap()
.unwrap();
assert!(status.sessions.is_empty());
assert_eq!(status.browsers[0].session_count, 0);
let _ = release_tx.send(());
responder.await.unwrap();
handle.shutdown().await;
}
#[tokio::test]
async fn timing_out_session_start_rolls_back_a_late_extension_success() {
let (handle, _sock) = spawn_daemon().await;
let state = handle.state();
let mut ws = connect_ext(handle.ws_addr()).await;
let _ = handshake_as_ext(&mut ws).await;
let (release_tx, release_rx) = tokio::sync::oneshot::channel();
let responder = tokio::spawn(async move {
respond_to_aborted_start(&mut ws, None, 4343).await;
let _ = release_rx.await;
});
let start = bsk::daemon::sessions::start_session(
&state.browsers,
&state.sessions,
&state.tool_queues,
None,
bsk::daemon::sessions::AgentWindowOptions::default(),
Duration::ZERO,
Duration::from_millis(20),
None,
)
.await;
assert!(matches!(
start,
Err(bsk::daemon::sessions::StartSessionError::Timeout)
));
assert!(state.sessions.is_empty());
let _ = release_tx.send(());
responder.await.unwrap();
handle.shutdown().await;
}
#[tokio::test]
async fn cancelling_session_stop_keeps_the_session_retryable() {
const STOP_RPC_ID: &str = "session-stop-cancel";
let (handle, sock) = spawn_daemon().await;
let mut ws = connect_ext(handle.ws_addr()).await;
let _ = handshake_as_ext(&mut ws).await;
let (stop_seen_tx, stop_seen_rx) = tokio::sync::oneshot::channel();
let (release_tx, release_rx) = tokio::sync::oneshot::channel();
let responder = tokio::spawn(async move {
respond_to_aborted_stop(&mut ws, Some(stop_seen_tx)).await;
let _ = release_rx.await;
});
let mut start_ipc = IpcClient::connect(&sock).await.unwrap();
let start: serde_json::Value = start_ipc
.call(
"start-before-stop-cancel",
Method::SessionStart,
Some(serde_json::json!({ "focused": false })),
Duration::from_secs(5),
)
.await
.unwrap()
.unwrap();
let session_id = start["session_id"].as_str().unwrap().to_string();
let mut stop_ipc = IpcClient::connect(&sock).await.unwrap();
let stop_session_id = session_id.clone();
let stop_call = tokio::spawn(async move {
let outcome: Result<serde_json::Value, RpcError> = stop_ipc
.call_with_id(
STOP_RPC_ID.to_string(),
Method::SessionStop,
Some(serde_json::json!({
"session_id": stop_session_id,
"all": false,
})),
Duration::from_secs(5),
)
.await
.unwrap();
outcome
});
stop_seen_rx.await.unwrap();
let mut cancel_ipc = IpcClient::connect(&sock).await.unwrap();
let cancel: CancelResult = cancel_ipc
.call(
"cancel-stop",
Method::Cancel,
Some(CancelParams {
rpc_id: STOP_RPC_ID.to_string(),
}),
Duration::from_secs(2),
)
.await
.unwrap()
.unwrap();
assert!(cancel.cancelled);
let stop_error = stop_call.await.unwrap().unwrap_err();
assert_eq!(stop_error.code, ErrorCode::Cancelled);
let mut status_ipc = IpcClient::connect(&sock).await.unwrap();
let status: StatusResult = status_ipc
.call::<(), _>(
"status-after-stop-cancel",
Method::SystemStatus,
None,
Duration::from_secs(2),
)
.await
.unwrap()
.unwrap();
assert_eq!(status.sessions.len(), 1);
assert_eq!(status.sessions[0].session_id, session_id);
let _ = release_tx.send(());
responder.await.unwrap();
handle.shutdown().await;
}
#[tokio::test]
async fn timing_out_session_stop_cancels_extension_teardown_and_keeps_session() {
let (handle, sock) = spawn_daemon().await;
let state = handle.state();
let mut ws = connect_ext(handle.ws_addr()).await;
let _ = handshake_as_ext(&mut ws).await;
let (release_tx, release_rx) = tokio::sync::oneshot::channel();
let responder = tokio::spawn(async move {
respond_to_aborted_stop(&mut ws, None).await;
let _ = release_rx.await;
});
let mut start_ipc = IpcClient::connect(&sock).await.unwrap();
let start: serde_json::Value = start_ipc
.call(
"start-before-stop-timeout",
Method::SessionStart,
Some(serde_json::json!({ "focused": false })),
Duration::from_secs(5),
)
.await
.unwrap()
.unwrap();
let session_id =
bsk::daemon::sessions::SessionId(start["session_id"].as_str().unwrap().to_string());
let stop = bsk::daemon::sessions::stop_session(
&state.browsers,
&state.sessions,
&state.tool_queues,
&state.session_interrupts,
&session_id,
Duration::from_millis(20),
None,
)
.await;
assert!(matches!(
stop,
Err(bsk::daemon::sessions::StopSessionError::Timeout)
));
assert!(state.sessions.get(&session_id).is_some());
let _ = release_tx.send(());
responder.await.unwrap();
handle.shutdown().await;
}
#[tokio::test]
async fn session_idle_timeout_stops_and_unregisters_session() {
let (handle, _sock) = spawn_daemon_with_session_idle(Duration::from_millis(50)).await;
crates/bsk-cli/tests/status_cmd.rs
+1 -1
View File
@@ -61,7 +61,7 @@ fn bsk_status_json_returns_structured_payload() {
assert!(parsed["pid"].as_u64().unwrap() > 0);
assert!(!parsed["daemon_version"].as_str().unwrap().is_empty());
assert_eq!(parsed["protocol_version"], "1.0");
assert_eq!(parsed["protocol_version"], "1.1");
assert!(parsed["sock_path"].as_str().is_some());
assert_eq!(parsed["browsers"], serde_json::json!([]));
assert_eq!(parsed["sessions"], serde_json::json!([]));
crates/bsk-cli/tests/tools_ipc.rs
+2
View File
@@ -478,6 +478,7 @@ async fn observe_returns_semantic_text_and_ref_count() {
tab_id: 13,
truncated: false,
dialogs: vec![],
hover_probe: None,
debug: None,
})
.unwrap(),
@@ -494,6 +495,7 @@ async fn observe_returns_semantic_text_and_ref_count() {
max_depth: None,
max_tokens: None,
debug_surfaces: false,
probe_hover: false,
},
)
.await
crates/bsk-cli/tests/tools_m7_ipc.rs
+92 -4
View File
@@ -12,11 +12,13 @@ use bsk::daemon::{self, DaemonConfig};
use bsk::ipc_client::IpcClient;
use bsk_protocol::system::{HandshakeParams, HandshakeResult};
use bsk_protocol::tools::{
ClickParams, ClickResult, FillParams, FillResult, KeyModifier, MouseButton, NavigateBackParams,
NavigateBackResult, NavigateForwardParams, NavigateForwardResult, NavigateParams,
NavigateResult, PressParams, PressResult, ReloadParams, ReloadResult, ScrollToParams,
ScrollToResult, SelectParams, SelectResult, SessionStartParams, SessionStartResult, WaitUntil,
BlurParams, BlurResult, ClickParams, ClickResult, FillParams, FillResult, FocusParams,
FocusResult, KeyModifier, MouseButton, NavigateBackParams, NavigateBackResult,
NavigateForwardParams, NavigateForwardResult, NavigateParams, NavigateResult, PressParams,
PressResult, ReloadParams, ReloadResult, SelectParams, SelectResult, SessionStartParams,
SessionStartResult, WaitUntil,
};
use bsk_protocol::tools::{ScrollToParams, ScrollToResult};
use bsk_protocol::{
BrowserPeerInfo, ErrorCode, Frame, Method, RequestFrame, ResponseBody, ResponseFrame, RpcError,
};
@@ -381,6 +383,92 @@ async fn click_round_trips_ref_and_modifiers() {
handle.shutdown().await;
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn focus_round_trips_verified_state() {
let (handle, sock) = spawn_daemon().await;
let mut ws = connect_ext(handle.ws_addr()).await;
let _ = do_handshake(&mut ws).await;
run_extension(ws, |req| {
assert_eq!(req.method, Method::ToolFocus);
let params = req.params.clone().unwrap();
let p: FocusParams = serde_json::from_value(params).unwrap();
assert_eq!(p.ref_.as_deref(), Some("@e3"));
ResponseBody::Ok(
serde_json::to_value(FocusResult {
tab_id: 9,
used_ref: Some("e3".into()),
used_selector: None,
focused: true,
dialogs: vec![],
})
.unwrap(),
)
});
let session_id = ipc_session_start(&sock).await;
let result: FocusResult = ipc_tool_call(
&sock,
Method::ToolFocus,
FocusParams {
session_id,
ref_: Some("@e3".into()),
selector: None,
tab_id: None,
timeout_ms: Some(5_000),
},
)
.await
.expect("focus ok");
assert!(result.focused);
assert_eq!(result.used_ref.as_deref(), Some("e3"));
handle.shutdown().await;
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn blur_round_trips_previous_focus_state() {
let (handle, sock) = spawn_daemon().await;
let mut ws = connect_ext(handle.ws_addr()).await;
let _ = do_handshake(&mut ws).await;
run_extension(ws, |req| {
assert_eq!(req.method, Method::ToolBlur);
let params = req.params.clone().unwrap();
let p: BlurParams = serde_json::from_value(params).unwrap();
assert_eq!(p.selector.as_deref(), Some("#search"));
ResponseBody::Ok(
serde_json::to_value(BlurResult {
tab_id: 9,
used_ref: None,
used_selector: Some("#search".into()),
was_focused: true,
focused: false,
dialogs: vec![],
})
.unwrap(),
)
});
let session_id = ipc_session_start(&sock).await;
let result: BlurResult = ipc_tool_call(
&sock,
Method::ToolBlur,
BlurParams {
session_id,
ref_: None,
selector: Some("#search".into()),
tab_id: None,
timeout_ms: Some(5_000),
},
)
.await
.expect("blur ok");
assert!(result.was_focused);
assert!(!result.focused);
assert_eq!(result.used_selector.as_deref(), Some("#search"));
handle.shutdown().await;
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn scroll_to_round_trips_visible_bounds() {
let (handle, sock) = spawn_daemon().await;
crates/bsk-cli/tests/windows_parent_cancel.rs
+342
View File
@@ -0,0 +1,342 @@
//! Real Windows CLI cancellation, including caller-owned transfer cleanup.
#![cfg(windows)]
use std::path::Path;
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use bsk::daemon::{file_transfer::TransferRegistry, info::DaemonInfo};
use bsk_protocol::tools::{TransferBeginParams, TransferIdParams};
use bsk_protocol::{ErrorCode, Frame, Method, ResponseBody, ResponseFrame, RpcError};
use serde_json::json;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::windows::named_pipe::ServerOptions;
use tokio::process::Command;
use tokio::sync::{Notify, watch};
fn response<T: serde::Serialize>(result: Result<T, RpcError>) -> ResponseBody {
match result {
Ok(value) => ResponseBody::Ok(serde_json::to_value(value).unwrap()),
Err(err) => ResponseBody::Err(err),
}
}
#[test]
fn stdin_close_cancels_business_but_releases_resources_with_a_bounded_budget() {
// Resolve the production registry's root before starting any runtime threads.
let registry_home = tempfile::tempdir().unwrap();
let original_home = std::env::var_os("BSK_HOME");
unsafe {
std::env::set_var("BSK_HOME", registry_home.path());
}
let transfers = Arc::new(TransferRegistry::new().unwrap());
unsafe {
match original_home {
Some(value) => std::env::set_var("BSK_HOME", value),
None => std::env::remove_var("BSK_HOME"),
}
}
transfers.initialize().unwrap();
tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap()
.block_on(exercise(
transfers,
&registry_home.path().join("run/transfers"),
));
}
async fn exercise(transfers: Arc<TransferRegistry>, staging_root: &Path) {
for mode in [
"wait",
"upload",
"upload-multi",
"upload-before-dispatch",
"download",
"upload-multi-stall",
"ordinary",
] {
let home = tempfile::tempdir().unwrap();
let pipe_name = format!(r"\\.\pipe\bsk-parent-cancel-{}", uuid::Uuid::new_v4());
let info = DaemonInfo::now(std::process::id(), (&pipe_name).into(), 0, "0.2.0");
std::fs::write(
home.path().join("daemon.json"),
serde_json::to_vec(&info).unwrap(),
)
.unwrap();
let mut pipe = ServerOptions::new()
.first_pipe_instance(true)
.create(&pipe_name)
.unwrap();
let registered = Arc::new(Notify::new());
let pending = Arc::new(Mutex::new(None::<String>));
let ids = Arc::new(Mutex::new(Vec::<String>::new()));
let released = Arc::new(Mutex::new(Vec::<String>::new()));
let received = Arc::new(Mutex::new(Vec::<Method>::new()));
let download_id = if mode == "download" {
let staging = transfers.begin_download("fixture").unwrap();
let browser_dir = home.path().join("BrowserSkill").join(&staging.transfer_id);
std::fs::create_dir_all(&browser_dir).unwrap();
let file = browser_dir.join("download.txt");
std::fs::write(&file, b"download fixture").unwrap();
transfers
.import_download(&staging.transfer_id, &file)
.unwrap();
ids.lock().unwrap().push(staging.transfer_id.clone());
Some(staging.transfer_id)
} else {
None
};
let (cancel, cancelled) = watch::channel(false);
let ready = Arc::clone(&registered);
let name = pipe_name.clone();
let registry = Arc::clone(&transfers);
let server_ids = Arc::clone(&ids);
let server_released = Arc::clone(&released);
let server_received = Arc::clone(&received);
let server = tokio::spawn(async move {
let mut connections = tokio::task::JoinSet::new();
loop {
pipe.connect().await.unwrap();
let next = ServerOptions::new().create(&name).unwrap();
let connection = std::mem::replace(&mut pipe, next);
let ready = Arc::clone(&ready);
let pending = Arc::clone(&pending);
let cancel = cancel.clone();
let mut cancelled = cancelled.clone();
let name = name.clone();
let registry = Arc::clone(&registry);
let ids = Arc::clone(&server_ids);
let released = Arc::clone(&server_released);
let received = Arc::clone(&server_received);
let download_id = download_id.clone();
connections.spawn(async move {
let (read, mut write) = tokio::io::split(connection);
let mut reader = BufReader::new(read);
let mut line = String::new();
while reader.read_line(&mut line).await.unwrap_or(0) > 0 {
let Frame::Request(req) = serde_json::from_str::<Frame>(&line).unwrap()
else {
panic!("expected request");
};
line.clear();
received.lock().unwrap().push(req.method.clone());
let params = req.params.clone().unwrap_or_default();
let body = match req.method {
Method::SystemStatus => ResponseBody::Ok(json!({
"daemon_version": "0.2.0", "protocol_version": "1.1",
"pid": std::process::id(), "uptime_secs": 0,
"ws_port": 0, "sock_path": name,
"browsers": [], "sessions": [], "version_skew_browsers": []
})),
Method::TransferBegin => {
let begin = registry
.begin_upload(serde_json::from_value(params).unwrap())
.unwrap();
ids.lock().unwrap().push(begin.transfer_id.clone());
response(Ok(begin))
}
Method::ToolDownload => ResponseBody::Ok(json!({
"tab_id": 1, "suggested_filename": "download.txt",
"byte_size": 16, "transfer_id": download_id,
})),
Method::ToolWaitMs if mode == "ordinary" => {
ResponseBody::Ok(json!({"waited_ms": 1}))
}
Method::TransferChunk
if mode == "upload-before-dispatch"
|| (mode.contains("multi")
&& ids.lock().unwrap().len() == 1) =>
{
response(
registry.write_chunk(serde_json::from_value(params).unwrap()),
)
}
Method::TransferFinish if mode != "upload-before-dispatch" => response(
registry.finish_upload(serde_json::from_value(params).unwrap()),
),
Method::ToolWaitMs
| Method::TransferChunk
| Method::TransferRead
| Method::TransferFinish => {
*pending.lock().unwrap() = Some(req.id.clone());
ready.notify_one();
cancelled.wait_for(|value| *value).await.unwrap();
if mode == "upload-before-dispatch" {
// Finish won the race with cancellation, but tool.upload must
// not be dispatched and the CLI still owns the staging.
response(
registry
.finish_upload(serde_json::from_value(params).unwrap()),
)
} else {
ResponseBody::Err(RpcError {
code: ErrorCode::Cancelled,
message: "parent cancellation reached daemon".into(),
data: None,
})
}
}
Method::Cancel => {
let target = params["rpc_id"].as_str().unwrap();
assert_eq!(pending.lock().unwrap().as_deref(), Some(target));
cancel.send(true).unwrap();
ResponseBody::Ok(json!({"cancelled": true}))
}
Method::TransferRelease => {
let params: TransferIdParams =
serde_json::from_value(params).unwrap();
released.lock().unwrap().push(params.transfer_id.clone());
if mode.contains("stall") {
std::future::pending::<()>().await;
}
response(Ok(registry.release(params)))
}
other => panic!("unexpected request after cancellation: {other:?}"),
};
let response = Frame::Response(ResponseFrame { id: req.id, body });
let mut bytes = serde_json::to_vec(&response).unwrap();
bytes.push(b'\n');
if write.write_all(&bytes).await.is_err() {
break;
}
}
});
}
});
let mut cmd = Command::new(env!("CARGO_BIN_EXE_bsk"));
cmd.env("BSK_HOME", home.path())
.env("BSK_AUTO_UPDATE", "off")
.env_remove("BSK_CANCEL_ON_STDIN_CLOSE")
.stdout(std::process::Stdio::piped())
.stderr(std::process::Stdio::piped())
.creation_flags(0x0800_0000)
.kill_on_drop(true);
if mode.starts_with("upload") {
let source = home.path().join("upload.txt");
// Declare 300 MiB without allocating or transferring that much data.
std::fs::File::create(&source)
.unwrap()
.set_len(if mode == "upload-before-dispatch" {
16
} else {
300 * 1024 * 1024
})
.unwrap();
cmd.args(["upload", "--session", "fixture", "--selector", "input"]);
if mode.contains("multi") {
let first = home.path().join("first.txt");
std::fs::write(&first, b"first file").unwrap();
cmd.arg("--file").arg(first);
}
cmd.arg("--file").arg(source);
} else if mode == "download" {
let out = home.path().join("out.txt");
std::fs::write(&out, b"existing output").unwrap();
cmd.args([
"download",
"--session",
"fixture",
"--selector",
"a",
"--overwrite",
"--out",
])
.arg(out);
} else {
cmd.args(["wait-ms", if mode == "ordinary" { "1ms" } else { "60s" }]);
}
cmd.arg("--json");
if mode == "ordinary" {
cmd.stdin(std::process::Stdio::null());
} else {
cmd.env("BSK_CANCEL_ON_STDIN_CLOSE", "1")
.stdin(std::process::Stdio::piped());
}
let mut child = cmd.spawn().unwrap();
if mode != "ordinary" {
tokio::time::timeout(Duration::from_secs(10), registered.notified())
.await
.expect("CLI registered business RPC");
child
.stdin
.as_mut()
.unwrap()
.write_all(b"still connected")
.await
.unwrap();
drop(child.stdin.take());
}
let start = Instant::now();
let output = tokio::time::timeout(Duration::from_secs(8), child.wait_with_output())
.await
.expect("CLI must settle promptly, including cleanup")
.unwrap();
server.abort();
let _ = server.await;
assert_eq!(
output.status.code(),
Some(if mode == "ordinary" { 0 } else { 2 }),
"{mode}: stdout={} stderr={}",
String::from_utf8_lossy(&output.stdout),
String::from_utf8_lossy(&output.stderr)
);
if mode != "ordinary" {
let body: serde_json::Value = serde_json::from_slice(&output.stdout).unwrap();
assert_eq!(body["code"], "cancelled");
if mode != "upload-before-dispatch" {
assert_eq!(body["message"], "parent cancellation reached daemon");
}
let received = received.lock().unwrap();
let cancel = received.iter().position(|m| *m == Method::Cancel).unwrap();
assert!(
received[cancel + 1..]
.iter()
.all(|m| *m == Method::TransferRelease),
"only cleanup may follow cancellation: {received:?}"
);
}
let ids = ids.lock().unwrap();
if mode.contains("stall") {
assert_eq!(released.lock().unwrap().len(), 1);
assert!(start.elapsed() >= Duration::from_secs(4));
// The CLI cannot force an unresponsive daemon to clean up.
transfers.release_session("fixture");
} else {
assert_eq!(
*released.lock().unwrap(),
*ids,
"{mode}: every allocation must be released"
);
for id in ids.iter() {
assert!(!staging_root.join(id).exists(), "staging leaked: {id}");
}
if mode.starts_with("upload") {
let retry = transfers
.begin_upload(TransferBeginParams {
session_id: "fixture".into(),
name: "retry.txt".into(),
byte_size: 300 * 1024 * 1024,
})
.expect("same session must have room for another 300 MiB upload");
transfers.release(TransferIdParams {
transfer_id: retry.transfer_id,
});
}
}
if mode == "download" {
assert_eq!(
std::fs::read(home.path().join("out.txt")).unwrap(),
b"existing output"
);
assert!(
!std::fs::read_dir(home.path())
.unwrap()
.flatten()
.any(|entry| entry.file_name().to_string_lossy().ends_with(".part"))
);
}
}
}
crates/bsk-cli/tests/windows_update.rs
+343
View File
@@ -0,0 +1,343 @@
//! Exercise self-update with a real executable and a local release server.
#![cfg(windows)]
use std::fs;
use std::io::{Cursor, Read, Write};
use std::net::{TcpListener, TcpStream};
use std::os::windows::process::CommandExt;
use std::path::PathBuf;
use std::process::{Child, Command, Stdio};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::thread;
use std::time::{Duration, Instant};
use bsk::daemon::info::DaemonInfo;
use sha2::{Digest, Sha256};
const MARKER: &[u8] = b"windows-update-regression-fixture";
struct ReleaseServer {
url: String,
requests: Arc<AtomicUsize>,
stop: Arc<AtomicBool>,
worker: Option<thread::JoinHandle<()>>,
}
impl ReleaseServer {
fn new(binary: &[u8]) -> Self {
let mut archive = zip::ZipWriter::new(Cursor::new(Vec::new()));
archive
.start_file(
"bsk.exe",
zip::write::SimpleFileOptions::default()
.compression_method(zip::CompressionMethod::Stored),
)
.unwrap();
archive.write_all(binary).unwrap();
let archive = archive.finish().unwrap().into_inner();
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
listener.set_nonblocking(true).unwrap();
let url = format!("http://{}", listener.local_addr().unwrap());
let manifest = serde_json::to_vec(&serde_json::json!({
"version": "999.0.0",
"assets": {"windows-x64": {
"url": format!("{url}/bsk.zip"),
"sha256": Sha256::digest(&archive).iter().map(|byte| format!("{byte:02x}")).collect::<String>(),
}},
}))
.unwrap();
let stop = Arc::new(AtomicBool::new(false));
let worker_stop = Arc::clone(&stop);
let requests = Arc::new(AtomicUsize::new(0));
let worker_requests = Arc::clone(&requests);
let worker = thread::spawn(move || {
while !worker_stop.load(Ordering::SeqCst) {
let (mut stream, _) = match listener.accept() {
Ok(connection) => connection,
Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => {
thread::sleep(Duration::from_millis(10));
continue;
}
Err(err) => panic!("accept release request: {err}"),
};
// Windows accept() inherits the listener's nonblocking mode.
// Keep accept polling for shutdown, but read/write each request
// in blocking mode with the bounded timeouts below.
stream.set_nonblocking(false).unwrap();
stream
.set_read_timeout(Some(Duration::from_secs(5)))
.unwrap();
stream
.set_write_timeout(Some(Duration::from_secs(5)))
.unwrap();
let mut request = Vec::new();
let mut chunk = [0; 1024];
while !request.windows(4).any(|window| window == b"\r\n\r\n") {
let len = stream.read(&mut chunk).unwrap();
if len == 0 {
break;
}
request.extend_from_slice(&chunk[..len]);
}
let body = if request.starts_with(b"GET /bsk.zip ") {
worker_requests.fetch_add(1, Ordering::SeqCst);
&archive
} else {
&manifest
};
write!(
stream,
"HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
body.len()
)
.unwrap();
stream.write_all(body).unwrap();
}
});
Self {
url: format!("{url}/version.json"),
requests,
stop,
worker: Some(worker),
}
}
}
impl Drop for ReleaseServer {
fn drop(&mut self) {
self.stop.store(true, Ordering::SeqCst);
let _ = self.worker.take().unwrap().join();
}
}
#[test]
fn release_server_waits_for_delayed_and_fragmented_request_headers() {
let server = ReleaseServer::new(b"test binary");
let address = server
.url
.strip_prefix("http://")
.unwrap()
.strip_suffix("/version.json")
.unwrap();
let mut stream = TcpStream::connect(address).unwrap();
stream
.set_read_timeout(Some(Duration::from_secs(5)))
.unwrap();
stream
.set_write_timeout(Some(Duration::from_secs(5)))
.unwrap();
// Give accept() time to run before any bytes arrive, then split the
// headers across writes. Both reads must wait rather than return WouldBlock.
thread::sleep(Duration::from_millis(100));
stream.write_all(b"GET /version.json HTTP/1.1\r\n").unwrap();
thread::sleep(Duration::from_millis(100));
stream.write_all(b"Host: localhost\r\n\r\n").unwrap();
let mut response = String::new();
stream.read_to_string(&mut response).unwrap();
assert!(response.starts_with("HTTP/1.1 200 OK\r\n"), "{response}");
let (_, body) = response.split_once("\r\n\r\n").unwrap();
let manifest: serde_json::Value = serde_json::from_str(body).unwrap();
assert_eq!(manifest["version"], "999.0.0");
}
struct Fixture {
_tmp: tempfile::TempDir,
exe: PathBuf,
home: PathBuf,
binary: Vec<u8>,
server: ReleaseServer,
daemon: Option<Child>,
}
impl Fixture {
fn new() -> Self {
let tmp = tempfile::TempDir::new().unwrap();
let dir = tmp.path().join("中文 space %PATH% ! & (update)");
fs::create_dir(&dir).unwrap();
let exe = dir.join("bsk.exe");
let home = tmp.path().join("home");
fs::create_dir(&home).unwrap();
fs::copy(env!("CARGO_BIN_EXE_bsk"), &exe).unwrap();
// A PE overlay distinguishes the replacement without requiring a
// second build or changing the executable's behavior/version.
let mut binary = fs::read(&exe).unwrap();
binary.extend_from_slice(MARKER);
let server = ReleaseServer::new(&binary);
Self {
_tmp: tmp,
exe,
home,
binary,
server,
daemon: None,
}
}
fn command(&self) -> Command {
let mut cmd = Command::new(&self.exe);
cmd.env("BSK_HOME", &self.home)
.env("BSK_UPDATE_MANIFEST_URL", &self.server.url)
.env("BSK_AUTO_UPDATE", "off")
.env("RUST_LOG", "info")
.env("NO_PROXY", "127.0.0.1,localhost")
.env_remove("BSK_DAEMONIZED")
.env_remove("BSK_DAEMON_REPLACES_PID")
.stdin(Stdio::null())
.creation_flags(0x0800_0000);
cmd
}
fn wait_for(&self, description: &str, mut check: impl FnMut() -> bool) {
let deadline = Instant::now() + Duration::from_secs(30);
while Instant::now() < deadline {
if check() {
return;
}
thread::sleep(Duration::from_millis(50));
}
let diagnostics: Vec<_> = fs::read_dir(self.exe.parent().unwrap())
.unwrap()
.flatten()
.filter(|entry| entry.path().extension().is_some_and(|ext| ext == "log"))
.map(|entry| {
(
entry.path(),
String::from_utf8_lossy(&fs::read(entry.path()).unwrap_or_default())
.into_owned(),
)
})
.collect();
let home_logs: Vec<_> = fs::read_dir(&self.home)
.unwrap()
.flatten()
.filter(|entry| entry.file_name().to_string_lossy().contains("log"))
.map(|entry| {
(
entry.path(),
String::from_utf8_lossy(&fs::read(entry.path()).unwrap_or_default())
.into_owned(),
)
})
.collect();
panic!(
"timed out waiting for {description}; helper logs: {diagnostics:?}; daemon logs: {home_logs:?}"
);
}
fn info(&self) -> Option<DaemonInfo> {
serde_json::from_slice(&fs::read(self.home.join("daemon.json")).ok()?).ok()
}
fn updated(&self) -> bool {
// A locked executable may briefly reject reads during replacement.
fs::read(&self.exe).is_ok_and(|binary| binary == self.binary)
}
}
impl Drop for Fixture {
fn drop(&mut self) {
// Stop only the daemon belonging to this isolated BSK_HOME. Also reap
// the foreground process on assertion failures to avoid leaking locks.
let _ = self.command().args(["daemon", "stop"]).output();
if let Some(child) = self.daemon.as_mut() {
let _ = child.kill();
let _ = child.wait();
}
}
}
#[test]
fn manual_update_replaces_the_running_executable_after_cli_exit() {
let fixture = Fixture::new();
let out = fixture
.command()
.args(["--json", "update", "--yes"])
.output()
.unwrap();
assert!(
out.status.success(),
"{}",
String::from_utf8_lossy(&out.stderr)
);
let report: serde_json::Value = serde_json::from_slice(&out.stdout).unwrap();
assert_eq!(report["status"], "staged");
fixture.wait_for("CLI replacement and helper cleanup", || {
fixture.updated() && fs::read_dir(fixture.exe.parent().unwrap()).unwrap().count() == 1
});
assert!(
fixture.info().is_none(),
"an update must not start a previously absent daemon"
);
assert_eq!(fixture.server.requests.load(Ordering::SeqCst), 1);
}
#[test]
fn automatic_update_exits_old_daemon_and_restarts_on_the_same_port() {
let mut fixture = Fixture::new();
let port = unused_port();
let log = fs::File::create(fixture.home.join("foreground.log")).unwrap();
let child = fixture
.command()
.env("BSK_AUTO_UPDATE", "on")
// Exercise the environment inherited by a normally detached daemon.
.env("BSK_DAEMONIZED", "1")
.args([
"daemon",
"start",
"--port",
&port.to_string(),
"--session-idle",
"1234ms",
"--daemon-idle",
"30s",
])
.stdout(Stdio::from(log.try_clone().unwrap()))
.stderr(Stdio::from(log))
.spawn()
.unwrap();
let old_pid = child.id();
fixture.daemon = Some(child);
fixture.wait_for("replacement daemon", || {
fixture.updated()
&& fixture
.info()
.is_some_and(|info| info.pid != old_pid && info.ws_port == port)
});
assert!(
fixture
.daemon
.as_mut()
.unwrap()
.try_wait()
.unwrap()
.is_some(),
"old daemon must exit"
);
fixture.wait_for("helper cleanup", || {
fs::read_dir(fixture.exe.parent().unwrap()).unwrap().count() == 1
});
let status = fixture
.command()
.args(["--json", "status"])
.output()
.unwrap();
assert!(
status.status.success(),
"{}",
String::from_utf8_lossy(&status.stderr)
);
assert_eq!(
fixture.server.requests.load(Ordering::SeqCst),
1,
"the replacement must not stage another update immediately"
);
}
fn unused_port() -> u16 {
TcpListener::bind("127.0.0.1:0")
.unwrap()
.local_addr()
.unwrap()
.port()
}
crates/bsk-cli/tests/ws_handshake.rs
+2 -2
View File
@@ -53,7 +53,7 @@ pub async fn send_handshake(
let params = HandshakeParams {
client: "browser-skill-extension".into(),
version: "0.1.0-dev.0".parse().unwrap(),
protocol_version: "1.0".into(),
protocol_version: "1.1".into(),
instance_id: instance_id.into(),
browser: BrowserPeerInfo {
name: "chrome".into(),
@@ -91,7 +91,7 @@ async fn ws_handshake_registers_browser_in_state() {
let mut ws = connect_ext(handle.ws_addr(), &origin).await;
let result = send_handshake(&mut ws, TEST_EXT_ID).await;
assert_eq!(result.server, "browser-skill-daemon");
assert_eq!(result.protocol_version, "1.0");
assert_eq!(result.protocol_version, "1.1");
let state = handle.state();
let browsers = state.browsers.snapshot();