feat(cli): write record output as a trace directory bundle

Persist `trace.json` plus per-state page VOM files under `--output`, tolerate
raced session teardown on stop, and clarify version-mismatch recovery hints.
This commit is contained in:
“haonan”
2026-08-13 19:38:05 +08:00
parent 13a61b2688
commit 025751ef03
4 changed files with 905 additions and 52 deletions
+595 -49
View File
@@ -1,15 +1,17 @@
//! `bsk record start|stop` — capture user actions in the Agent Window.
use std::fs;
use std::path::PathBuf;
use std::collections::HashSet;
use std::fs::{self, OpenOptions};
use std::io;
use std::path::{Path, PathBuf};
use std::time::Duration;
use anyhow::Context;
use bsk_protocol::Method;
use bsk_protocol::tools::{
RecordAwaitParams, RecordAwaitResult, RecordStartParams, RecordStartResult, RecordStopParams,
RecordStopResult, Trace,
RecordStopResult, TRACE_VERSION, Trace, TraceState,
};
use bsk_protocol::{ErrorCode, Method};
use clap::{Args, Subcommand};
use crate::cli::TOOL_IPC_TIMEOUT;
@@ -18,10 +20,34 @@ use crate::cli::ensure_daemon::ensure_daemon;
use crate::cli::error::{CliError, Format};
use crate::cli::record_state;
use crate::cli::session::{SessionStartOptions, start_session, stop_session};
use crate::daemon::ipc::STOP_IN_PROGRESS_REASON;
/// Max wait for the user to click 结束 in the browser (24 hours).
const RECORD_AWAIT_TIMEOUT_MS: u32 = 86_400_000;
/// Tear down the recording session, tolerating the teardown the other half of
/// the pair already started. `record start` blocks in `record_await` and stops
/// the session once the trace arrives; the documented `record stop` fallback
/// stops it too. Whichever loses that race must not report a failure — the
/// trace is already written by then.
fn stop_recording_session(sock: PathBuf, session_id: &str) -> Result<(), CliError> {
match stop_session(sock, session_id) {
Ok(_) => Ok(()),
Err(err) if session_teardown_raced(&err) => Ok(()),
Err(err) => Err(err),
}
}
fn session_teardown_raced(err: &CliError) -> bool {
if err.code() == Some(ErrorCode::NotFound) {
return true;
}
err.data()
.and_then(|data| data.get("reason"))
.and_then(serde_json::Value::as_str)
== Some(STOP_IN_PROGRESS_REASON)
}
/// `bsk record …` subcommand tree.
#[derive(Debug, Clone, Args)]
pub struct RecordCmd {
@@ -58,15 +84,25 @@ pub struct RecordStartArgs {
#[arg(long)]
pub purpose: Option<String>,
/// Output path for the trace JSON (default `./trace.json`).
#[arg(long, default_value = "trace.json")]
/// Max VOM tokens per page observation file (default 3000).
#[arg(long = "max-page-tokens")]
pub max_page_tokens: Option<u32>,
/// Redact all form values in page observations (`[filled]` only).
#[arg(long = "redact-values")]
pub redact_values: bool,
/// Output directory for the trace bundle (default `./trace`).
/// Writes `<dir>/trace.json` and `<dir>/pages/*.vom.txt`.
#[arg(long, default_value = "trace")]
pub output: PathBuf,
}
#[derive(Debug, Clone, Args)]
pub struct RecordStopArgs {
/// Output path for the trace JSON (default `./trace.json`).
#[arg(long, default_value = "trace.json")]
/// Output directory for the trace bundle (default `./trace`).
/// Writes `<dir>/trace.json` and `<dir>/pages/*.vom.txt`.
#[arg(long, default_value = "trace")]
pub output: PathBuf,
}
@@ -98,6 +134,8 @@ fn dispatch_start(args: RecordStartArgs, format: Format) -> Result<(), CliError>
tab_id: args.tab_id,
url: args.url,
purpose: args.purpose.clone(),
max_page_tokens: args.max_page_tokens,
redact_values: Some(args.redact_values),
};
let start_result = business_rpc::call::<RecordStartParams, RecordStartResult>(
info.sock_path.clone(),
@@ -142,13 +180,13 @@ fn dispatch_start(args: RecordStartArgs, format: Format) -> Result<(), CliError>
// Keep write/render inside a Result so `?` cannot skip session teardown.
let run_result: Result<(), CliError> = match await_result {
Ok(result) => (|| {
write_trace_file(&args.output, &result.trace)?;
render_finish(&result.trace, &args.output, format)
let pages_dir = write_trace_bundle(&args.output, &result.trace)?;
render_finish(&result.trace, &args.output, &pages_dir, format)
})(),
Err(err) => Err(err),
};
let session_stop_result = stop_session(info.sock_path, &session.session_id);
let session_stop_result = stop_recording_session(info.sock_path, &session.session_id);
record_state::clear();
run_result?;
@@ -161,22 +199,23 @@ fn dispatch_stop(args: RecordStopArgs, format: Format) -> Result<(), CliError> {
let info = ensure_daemon().context("ensure daemon is running")?;
let session_id = state.session_id.clone();
let params = RecordStopParams {
session_id: session_id.clone(),
};
let result = business_rpc::call::<RecordStopParams, RecordStopResult>(
info.sock_path.clone(),
"record-stop",
Method::ToolRecordStop,
Some(params),
TOOL_IPC_TIMEOUT,
)?;
let run_result: Result<(), CliError> = (|| {
let params = RecordStopParams {
session_id: session_id.clone(),
};
let result = business_rpc::call::<RecordStopParams, RecordStopResult>(
info.sock_path.clone(),
"record-stop",
Method::ToolRecordStop,
Some(params),
TOOL_IPC_TIMEOUT,
)?;
write_trace_file(&args.output, &result.trace)?;
render_stop(&result, &args.output, format)
let pages_dir = write_trace_bundle(&args.output, &result.trace)?;
render_stop(&result, &args.output, &pages_dir, format)
})();
let session_stop_result = stop_session(info.sock_path, &session_id);
let session_stop_result = stop_recording_session(info.sock_path, &session_id);
record_state::clear();
run_result?;
@@ -190,30 +229,313 @@ fn record_await_ipc_timeout(timeout_ms: u32) -> Duration {
.unwrap_or(Duration::from_secs(u64::from(timeout_ms / 1_000) + 15))
}
fn write_trace_file(output: &PathBuf, trace: &Trace) -> Result<(), CliError> {
let json = serde_json::to_string_pretty(trace)
.context("serialize trace JSON")
.map_err(CliError::Local)?;
if let Some(parent) = output.parent() {
if !parent.as_os_str().is_empty() {
fs::create_dir_all(parent)
.with_context(|| format!("create output directory {}", parent.display()))
.map_err(CliError::Local)?;
fn pages_dir_for_output(output_dir: &Path) -> PathBuf {
output_dir.join("pages")
}
fn trace_json_path(output_dir: &Path) -> PathBuf {
output_dir.join("trace.json")
}
fn is_canonical_state_id(id: &str) -> bool {
let Some(number) = id.strip_prefix('s') else {
return false;
};
let mut digits = number.bytes();
matches!(digits.next(), Some(b'1'..=b'9')) && digits.all(|byte| byte.is_ascii_digit())
}
fn validate_bundle_directory(path: &Path, description: &str) -> Result<(), CliError> {
match fs::symlink_metadata(path) {
Ok(metadata) if metadata.file_type().is_symlink() => Err(CliError::Local(anyhow::anyhow!(
"{description} {} must not be a symlink",
path.display()
))),
Ok(metadata) if metadata.is_dir() => Ok(()),
Ok(_) => Err(CliError::Local(anyhow::anyhow!(
"{description} {} is not a directory",
path.display()
))),
Err(err) if err.kind() == io::ErrorKind::NotFound => Ok(()),
Err(err) => Err(CliError::Local(
anyhow::Error::new(err).context(format!("inspect {description} {}", path.display())),
)),
}
}
fn validate_replaceable_file(path: &Path, description: &str) -> Result<(), CliError> {
match fs::symlink_metadata(path) {
Ok(metadata) if metadata.file_type().is_file() => Ok(()),
Ok(_) => Err(CliError::Local(anyhow::anyhow!(
"{description} {} is not a regular file",
path.display()
))),
Err(err) if err.kind() == io::ErrorKind::NotFound => Ok(()),
Err(err) => Err(CliError::Local(
anyhow::Error::new(err).context(format!("inspect {description} {}", path.display())),
)),
}
}
fn commit_staged_file(staged: &Path, target: &Path, backup: &Path) -> io::Result<Option<PathBuf>> {
let old_file = match fs::symlink_metadata(target) {
Ok(_) => {
fs::rename(target, backup)?;
Some(backup.to_path_buf())
}
Err(err) if err.kind() == io::ErrorKind::NotFound => None,
Err(err) => return Err(err),
};
if let Err(err) = fs::rename(staged, target) {
if let Some(old_file) = old_file.as_deref() {
if let Err(restore_err) = fs::rename(old_file, target) {
return Err(io::Error::new(
restore_err.kind(),
format!(
"{err}; additionally failed to restore {}: {restore_err}",
target.display()
),
));
}
}
return Err(err);
}
Ok(old_file)
}
fn rollback_installed_files(installed: &[(PathBuf, Option<PathBuf>)]) -> io::Result<()> {
let mut rollback_error = None;
for (target, backup) in installed.iter().rev() {
match fs::remove_file(target) {
Ok(()) => {}
Err(err) if err.kind() == io::ErrorKind::NotFound => {}
Err(err) => {
rollback_error.get_or_insert(err);
continue;
}
}
if let Some(backup) = backup {
if let Err(err) = fs::rename(backup, target) {
rollback_error.get_or_insert(err);
}
}
}
match rollback_error {
Some(err) => Err(err),
None => Ok(()),
}
}
/// Drop page files left by an earlier, longer recording without deleting the
/// directory itself: `record start` and the `record stop` fallback export the
/// same recording concurrently, and a `remove_dir_all` here would fail the
/// writer whose files the other one just erased.
fn prune_stale_pages(pages_dir: &Path, keep: &HashSet<String>) -> Result<(), CliError> {
let entries = match fs::read_dir(pages_dir) {
Ok(entries) => entries,
Err(err) if err.kind() == io::ErrorKind::NotFound => return Ok(()),
Err(err) => {
return Err(CliError::Local(
anyhow::Error::new(err).context(format!("read pages dir {}", pages_dir.display())),
));
}
};
for entry in entries.flatten() {
let name = entry.file_name();
let Some(name) = name.to_str() else { continue };
if !name.ends_with(".vom.txt") || keep.contains(name) {
continue;
}
match fs::remove_file(entry.path()) {
Ok(()) => {}
Err(err) if err.kind() == io::ErrorKind::NotFound => {}
Err(err) => {
return Err(CliError::Local(
anyhow::Error::new(err)
.context(format!("remove stale page {}", entry.path().display())),
));
}
}
}
fs::write(output, format!("{json}\n"))
.with_context(|| format!("write trace to {}", output.display()))
.map_err(CliError::Local)?;
Ok(())
}
fn render_finish(trace: &Trace, output: &PathBuf, format: Format) -> Result<(), CliError> {
fn write_trace_bundle(output_dir: &Path, trace: &Trace) -> Result<PathBuf, CliError> {
let pages_dir = pages_dir_for_output(output_dir);
let trace_path = trace_json_path(output_dir);
if trace.version != TRACE_VERSION {
return Err(CliError::Local(anyhow::anyhow!(
"trace version {} is not supported (expected {})",
trace.version,
TRACE_VERSION
)));
}
let mut state_ids = HashSet::with_capacity(trace.states.len());
for state in &trace.states {
if !is_canonical_state_id(&state.id) {
return Err(CliError::Local(anyhow::anyhow!(
"state id {:?} is not canonical (expected s1, s2, ...)",
state.id
)));
}
if !state_ids.insert(state.id.as_str()) {
return Err(CliError::Local(anyhow::anyhow!(
"duplicate state id {:?}",
state.id
)));
}
}
let mut disk_states: Vec<TraceState> = Vec::with_capacity(trace.states.len());
let mut page_writes = Vec::new();
let mut keep = HashSet::new();
for state in &trace.states {
let mut disk = state.clone();
if let Some(body) = state.body.as_deref() {
let filename = format!("{}.vom.txt", state.id);
let page_path = pages_dir.join(&filename);
if page_path.parent() != Some(pages_dir.as_path()) {
return Err(CliError::Local(anyhow::anyhow!(
"page path {} escapes pages directory {}",
page_path.display(),
pages_dir.display()
)));
}
keep.insert(filename.clone());
page_writes.push((filename.clone(), body));
disk.body = None;
disk.page = Some(filename);
}
disk_states.push(disk);
}
let mut disk_trace = trace.clone();
disk_trace.states = disk_states;
let json = serde_json::to_string_pretty(&disk_trace)
.context("serialize trace JSON")
.map_err(CliError::Local)?;
validate_bundle_directory(output_dir, "output bundle directory")?;
fs::create_dir_all(output_dir)
.with_context(|| format!("create output dir {}", output_dir.display()))
.map_err(CliError::Local)?;
validate_bundle_directory(&pages_dir, "pages directory")?;
// `record start` and `record stop` can export the same recording at once.
// Serialize writers without replacing the shared directory underneath one
// another; the lock is released when this file handle is dropped.
let lock_path = output_dir.join(".bsk-record-export.lock");
let lock_file = OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(false)
.open(&lock_path)
.with_context(|| format!("open export lock {}", lock_path.display()))
.map_err(CliError::Local)?;
fs2::FileExt::lock_exclusive(&lock_file)
.with_context(|| format!("lock export bundle {}", output_dir.display()))
.map_err(CliError::Local)?;
fs::create_dir_all(&pages_dir)
.with_context(|| format!("create pages dir {}", pages_dir.display()))
.map_err(CliError::Local)?;
for (filename, _) in &page_writes {
validate_replaceable_file(&pages_dir.join(filename), "page observation")?;
}
validate_replaceable_file(&trace_path, "trace JSON")?;
let transaction_id = uuid::Uuid::new_v4();
let staging_dir = output_dir.join(format!(".bsk-record-stage-{transaction_id}"));
let staging_pages = staging_dir.join("pages");
fs::create_dir_all(&staging_pages)
.with_context(|| format!("create staging dir {}", staging_pages.display()))
.map_err(CliError::Local)?;
let stage_result: Result<(), CliError> = (|| {
for (filename, body) in &page_writes {
let staged_path = staging_pages.join(filename);
fs::write(&staged_path, body)
.with_context(|| format!("write staged page observation {}", staged_path.display()))
.map_err(CliError::Local)?;
}
let staged_trace = staging_dir.join("trace.json");
fs::write(&staged_trace, format!("{json}\n"))
.with_context(|| format!("write staged trace to {}", staged_trace.display()))
.map_err(CliError::Local)?;
Ok(())
})();
if let Err(err) = stage_result {
let _ = fs::remove_dir_all(&staging_dir);
return Err(err);
}
let mut replacements: Vec<(PathBuf, PathBuf, PathBuf)> =
Vec::with_capacity(page_writes.len() + 1);
for (filename, _) in &page_writes {
replacements.push((
staging_pages.join(filename),
pages_dir.join(filename),
pages_dir.join(format!(".{filename}.{transaction_id}.bak")),
));
}
replacements.push((
staging_dir.join("trace.json"),
trace_path.clone(),
output_dir.join(format!(".trace.json.{transaction_id}.bak")),
));
let mut installed = Vec::with_capacity(replacements.len());
for (staged, target, backup) in replacements {
match commit_staged_file(&staged, &target, &backup) {
Ok(old_file) => installed.push((target, old_file)),
Err(err) => {
let rollback_result = rollback_installed_files(&installed);
let _ = fs::remove_dir_all(&staging_dir);
let message = match rollback_result {
Ok(()) => format!("commit replacement {}: {err}", target.display()),
Err(rollback_err) => format!(
"commit replacement {}: {err}; rollback also failed: {rollback_err}",
target.display()
),
};
return Err(CliError::Local(anyhow::anyhow!(message)));
}
}
}
for (_, backup) in &installed {
if let Some(backup) = backup {
let _ = fs::remove_file(backup);
}
}
let _ = fs::remove_dir_all(&staging_dir);
prune_stale_pages(&pages_dir, &keep)?;
Ok(pages_dir)
}
fn render_finish(
trace: &Trace,
output_dir: &PathBuf,
pages_dir: &PathBuf,
format: Format,
) -> Result<(), CliError> {
let trace_path = trace_json_path(output_dir);
match format {
Format::Json => {
println!(
"{}",
serde_json::to_string(&serde_json::json!({
"output": output,
"output": output_dir,
"trace_json": trace_path,
"pages_dir": pages_dir,
"trace": trace,
"window_closed": true,
}))
@@ -221,7 +543,13 @@ fn render_finish(trace: &Trace, output: &PathBuf, format: Format) -> Result<(),
);
}
Format::Human => {
println!("saved {} steps to {}", trace.steps.len(), output.display());
println!(
"saved {} steps and {} states to {} (+ {})",
trace.steps.len(),
trace.states.len(),
trace_path.display(),
pages_dir.display()
);
}
}
Ok(())
@@ -230,33 +558,251 @@ fn render_finish(trace: &Trace, output: &PathBuf, format: Format) -> Result<(),
fn render_stop(
result: &RecordStopResult,
output: &PathBuf,
pages_dir: &PathBuf,
format: Format,
) -> Result<(), CliError> {
render_finish(&result.trace, output, format)
render_finish(&result.trace, output, pages_dir, format)
}
#[cfg(test)]
mod tests {
use super::*;
use bsk_protocol::tools::{
NavigationCause, RecorderInfo, Step, StepCommon, StepResult, StopReason, TraceEntry,
VOM_FORMAT_VERSION,
};
#[test]
fn default_output_is_trace_json() {
let args = RecordStopArgs {
output: PathBuf::from("trace.json"),
};
assert_eq!(args.output, PathBuf::from("trace.json"));
fn sample_trace(state_id: &str, body: &str) -> Trace {
Trace {
version: TRACE_VERSION,
recorded_at: "2026-08-10T02:12:55.360Z".into(),
stopped_by: StopReason::UserFinish,
entry: TraceEntry {
start_url: "https://example.com/".into(),
},
recorder: RecorderInfo {
bsk: "0.1.10".into(),
vom: VOM_FORMAT_VERSION,
},
states: vec![TraceState {
id: state_id.into(),
url: "https://example.com/".into(),
title: Some("Example".into()),
body: Some(body.into()),
page: None,
truncated: false,
}],
steps: vec![Step::Navigate {
common: StepCommon {
id: 1,
state: state_id.into(),
result: StepResult {
state: state_id.into(),
},
},
to: "https://example.com/".into(),
cause: NavigationCause::UserTyped,
}],
purpose: None,
started_at: None,
}
}
#[test]
fn start_args_default_output_is_trace_json() {
fn concurrent_teardown_is_not_a_record_failure() {
let stopping = CliError::from_rpc(bsk_protocol::RpcError {
code: ErrorCode::Timeout,
message: "session stop is already in progress".into(),
data: Some(serde_json::json!({ "reason": STOP_IN_PROGRESS_REASON })),
});
assert!(session_teardown_raced(&stopping));
let gone = CliError::from_rpc(bsk_protocol::RpcError {
code: ErrorCode::NotFound,
message: "session is not registered".into(),
data: None,
});
assert!(session_teardown_raced(&gone));
}
#[test]
fn unrelated_stop_timeout_still_fails() {
let stalled = CliError::from_rpc(bsk_protocol::RpcError {
code: ErrorCode::Timeout,
message: "tool.session_stop timed out".into(),
data: None,
});
assert!(!session_teardown_raced(&stalled));
}
#[test]
fn default_output_is_trace_dir() {
let args = RecordStopArgs {
output: PathBuf::from("trace"),
};
assert_eq!(args.output, PathBuf::from("trace"));
}
#[test]
fn start_args_default_output_is_trace_dir() {
let args = RecordStartArgs {
browser: None,
tab_id: None,
url: None,
purpose: None,
output: PathBuf::from("trace.json"),
max_page_tokens: None,
redact_values: false,
output: PathBuf::from("trace"),
};
assert_eq!(args.output, PathBuf::from("trace.json"));
assert_eq!(args.output, PathBuf::from("trace"));
}
#[test]
fn write_trace_bundle_writes_dir_layout() {
let dir = tempfile::tempdir().unwrap();
let output = dir.path().join("flow");
let trace = sample_trace("s1", "@vom 1\nL1 page");
let pages_dir = write_trace_bundle(&output, &trace).unwrap();
let trace_path = output.join("trace.json");
assert!(trace_path.exists());
assert_eq!(pages_dir, output.join("pages"));
assert!(pages_dir.join("s1.vom.txt").exists());
let disk: serde_json::Value =
serde_json::from_str(&fs::read_to_string(&trace_path).unwrap()).unwrap();
assert_eq!(disk["states"][0]["page"], "s1.vom.txt");
assert!(disk["states"][0].get("body").is_none());
assert_eq!(
fs::read_to_string(pages_dir.join("s1.vom.txt")).unwrap(),
"@vom 1\nL1 page"
);
// Re-exporting drops pages from the previous run but keeps the
// directory in place for the concurrent writer.
fs::write(pages_dir.join("s9.vom.txt"), "stale").unwrap();
fs::write(pages_dir.join("notes.md"), "keep me").unwrap();
write_trace_bundle(&output, &trace).unwrap();
assert!(!pages_dir.join("s9.vom.txt").exists());
assert!(pages_dir.join("s1.vom.txt").exists());
assert!(pages_dir.join("notes.md").exists());
}
#[test]
fn write_trace_bundle_rejects_traversal_state_id() {
let dir = tempfile::tempdir().unwrap();
let output = dir.path().join("flow");
let trace = sample_trace("../escaped", "@vom 1\nL1 page");
let err = write_trace_bundle(&output, &trace).unwrap_err();
assert!(err.to_string().contains("state id"));
assert!(!output.join("escaped.vom.txt").exists());
assert!(!output.join("trace.json").exists());
}
#[test]
fn write_trace_bundle_rejects_duplicate_state_id() {
let dir = tempfile::tempdir().unwrap();
let output = dir.path().join("flow");
let mut trace = sample_trace("s1", "@vom 1\nL1 first");
trace.states.push(TraceState {
body: Some("@vom 1\nL1 duplicate".into()),
..trace.states[0].clone()
});
let err = write_trace_bundle(&output, &trace).unwrap_err();
assert!(err.to_string().contains("duplicate state id"));
assert!(!output.exists());
}
#[test]
fn unsupported_version_preserves_existing_bundle() {
let dir = tempfile::tempdir().unwrap();
let output = dir.path().join("flow");
write_trace_bundle(&output, &sample_trace("s1", "original")).unwrap();
let original_json = fs::read(output.join("trace.json")).unwrap();
let mut replacement = sample_trace("s2", "replacement");
replacement.version += 1;
let err = write_trace_bundle(&output, &replacement).unwrap_err();
assert!(err.to_string().contains("not supported"));
assert_eq!(fs::read(output.join("trace.json")).unwrap(), original_json);
assert_eq!(
fs::read_to_string(output.join("pages/s1.vom.txt")).unwrap(),
"original"
);
assert!(!output.join("pages/s2.vom.txt").exists());
}
#[test]
fn page_replacement_failure_preserves_existing_bundle() {
let dir = tempfile::tempdir().unwrap();
let output = dir.path().join("flow");
write_trace_bundle(&output, &sample_trace("s1", "original")).unwrap();
let original_json = fs::read(output.join("trace.json")).unwrap();
fs::write(output.join("pages/notes.md"), "keep me").unwrap();
fs::create_dir(output.join("pages/s2.vom.txt")).unwrap();
let err = write_trace_bundle(&output, &sample_trace("s2", "replacement")).unwrap_err();
assert!(err.to_string().contains("page"));
assert_eq!(fs::read(output.join("trace.json")).unwrap(), original_json);
assert_eq!(
fs::read_to_string(output.join("pages/s1.vom.txt")).unwrap(),
"original"
);
assert_eq!(
fs::read_to_string(output.join("pages/notes.md")).unwrap(),
"keep me"
);
}
#[cfg(unix)]
#[test]
fn write_trace_bundle_rejects_symlinked_pages_directory() {
use std::os::unix::fs::symlink;
let dir = tempfile::tempdir().unwrap();
let output = dir.path().join("flow");
let external_pages = dir.path().join("external-pages");
fs::create_dir_all(&output).unwrap();
fs::create_dir_all(&external_pages).unwrap();
fs::write(external_pages.join("s9.vom.txt"), "external").unwrap();
symlink(&external_pages, output.join("pages")).unwrap();
let err = write_trace_bundle(&output, &sample_trace("s1", "replacement")).unwrap_err();
assert!(err.to_string().contains("symlink"));
assert_eq!(
fs::read_to_string(external_pages.join("s9.vom.txt")).unwrap(),
"external"
);
assert!(!external_pages.join("s1.vom.txt").exists());
assert!(!output.join("trace.json").exists());
}
#[cfg(unix)]
#[test]
fn write_trace_bundle_rejects_symlinked_output_directory() {
use std::os::unix::fs::symlink;
let dir = tempfile::tempdir().unwrap();
let external_output = dir.path().join("external-output");
let output = dir.path().join("flow");
fs::create_dir_all(&external_output).unwrap();
fs::write(external_output.join("notes.md"), "external").unwrap();
symlink(&external_output, &output).unwrap();
let err = write_trace_bundle(&output, &sample_trace("s1", "replacement")).unwrap_err();
assert!(err.to_string().contains("symlink"));
assert_eq!(
fs::read_to_string(external_output.join("notes.md")).unwrap(),
"external"
);
assert!(!external_output.join("trace.json").exists());
assert!(!external_output.join("pages").exists());
}
#[test]
+27 -3
View File
@@ -141,9 +141,9 @@ pub fn info_for(code: ErrorCode) -> RenderInfo {
exit_code: 2,
},
ErrorCode::VersionTooOld => RenderInfo {
summary: "peer version is too old to communicate with this build",
summary: "the installed bsk CLI and BrowserSkill extension are incompatible",
hint: Some(
"upgrade both bsk CLI and the browser-skill extension so both sides satisfy `min_compatible_protocol`",
"update bsk and the BrowserSkill extension, reload the extension, then run `bsk daemon restart`",
),
exit_code: 5,
},
@@ -157,7 +157,7 @@ pub fn info_for(code: ErrorCode) -> RenderInfo {
ErrorCode::NoBrowserConnected => RenderInfo {
summary: "no browser is connected to the daemon",
hint: Some(
"open the browser-skill extension in your browser and wait for the popup to show \"connected\"",
"open the BrowserSkill extension and wait for it to show \"connected\"; if the extension is already open, update and reload it, then run `bsk daemon restart`",
),
exit_code: 1,
},
@@ -310,6 +310,30 @@ mod tests {
assert_eq!(exit_code_for(ErrorCode::VersionTooOld), 5);
}
#[test]
fn version_error_names_the_components_and_gives_update_steps() {
let info = info_for(ErrorCode::VersionTooOld);
assert!(info.summary.contains("bsk CLI"));
assert!(info.summary.contains("BrowserSkill extension"));
let hint = info
.hint
.expect("version mismatch should have a repair hint");
assert!(hint.contains("update bsk"));
assert!(hint.contains("reload the extension"));
assert!(hint.contains("bsk daemon restart"));
assert!(!hint.contains("min_compatible_protocol"));
}
#[test]
fn no_browser_hint_explains_version_mismatch_recovery() {
let hint = info_for(ErrorCode::NoBrowserConnected)
.hint
.expect("no browser should have a repair hint");
assert!(hint.contains("extension is already open"));
assert!(hint.contains("update"));
assert!(hint.contains("bsk daemon restart"));
}
/// Every code must carry a non-empty summary; an empty `summary`
/// would render `error: ` with nothing after it.
#[test]
+12
View File
@@ -786,7 +786,19 @@ async fn handle_session_stop(state: &Arc<DaemonState>, params: Value) -> Result<
}
}
/// Marks a `session.stop` rejected because another teardown already owns the
/// session, so callers that race on purpose (`bsk record`) can tell it apart
/// from a browser that stopped responding.
pub const STOP_IN_PROGRESS_REASON: &str = "session_stop_in_progress";
fn map_stop_error(err: StopSessionError) -> RpcError {
if matches!(err, StopSessionError::Stopping) {
return RpcError {
code: ErrorCode::Timeout,
message: err.to_string(),
data: Some(serde_json::json!({ "reason": STOP_IN_PROGRESS_REASON })),
};
}
let code = match &err {
StopSessionError::NotFound => ErrorCode::NotFound,
StopSessionError::Stopping => ErrorCode::Timeout,
+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, RecorderInfo, SessionStartParams, SessionStartResult, StopReason, Trace,
TraceEntry, TraceState, 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() -> Trace {
Trace {
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![TraceState {
id: "s1".into(),
url: "https://example.com/".into(),
title: Some("Example".into()),
body: Some("@vom 1\nL1 page".into()),
page: None,
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: 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() {
let temp = tempfile::Builder::new()
.prefix(".record-stop-retry-")
.tempdir_in(env!("CARGO_MANIFEST_DIR"))
.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("pages/s1.vom.txt").exists());
unsafe {
std::env::remove_var(BSK_HOME_ENV);
}
daemon.shutdown().await;
}