mirror of
https://github.com/Tencent/BrowserSkill.git
synced 2026-09-10 13:01:23 +08:00
merge: integrate main into focus and blur PR
This commit is contained in:
drakezhang
@@ -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"]);
|
||||
@@ -231,6 +280,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(&[
|
||||
|
||||
@@ -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"),
|
||||
|
||||
@@ -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");
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
@@ -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;
|
||||
}
|
||||
@@ -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;
|
||||
|
||||
@@ -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!([]));
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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,
|
||||
®istry_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(®istered);
|
||||
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(®istry);
|
||||
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"))
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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()
|
||||
}
|
||||
@@ -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();
|
||||
|
||||
Reference in New Issue
Block a user