mirror of
https://github.com/Tencent/BrowserSkill.git
synced 2026-09-10 13:01:23 +08:00
feat: add bsk record for semantic user-action traces
Capture user actions in the Agent Window and export a record-only trace (pages[] + navigate/click/fill/select/press steps with target descriptors). Adds the record overlay UI aligned with the control overlay, and wires record start/stop/await through CLI, daemon, protocol, and extension.
This commit is contained in:
@@ -19,6 +19,8 @@ pub mod interaction;
|
||||
pub mod logs;
|
||||
pub mod navigate;
|
||||
pub mod network;
|
||||
pub mod record;
|
||||
pub mod record_state;
|
||||
pub mod render_error;
|
||||
pub mod screenshot;
|
||||
pub mod session;
|
||||
@@ -39,6 +41,7 @@ use crate::cli::install_skill::InstallSkillArgs;
|
||||
use crate::cli::interaction::{ClickArgs, FillArgs, PressArgs, SelectArgs};
|
||||
use crate::cli::navigate::{NavigateCommand, NavigateHistoryArgs, ReloadArgs};
|
||||
use crate::cli::network::NetworkArgs;
|
||||
use crate::cli::record::RecordCmd;
|
||||
use crate::cli::screenshot::ScreenshotArgs;
|
||||
use crate::cli::session::SessionCmd;
|
||||
use crate::cli::snapshot::SnapshotArgs;
|
||||
@@ -169,6 +172,9 @@ pub enum Command {
|
||||
/// Ask the human to complete an in-page step (captcha / login / confirm).
|
||||
#[command(name = "request-help")]
|
||||
RequestHelp(RequestHelpArgs),
|
||||
|
||||
/// Record user actions into a semantic `trace.json` textbook for LLMs.
|
||||
Record(RecordCmd),
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Args, Default)]
|
||||
|
||||
@@ -0,0 +1,256 @@
|
||||
//! `bsk record start|stop` — capture user actions in the Agent Window.
|
||||
|
||||
use std::fs;
|
||||
use std::path::PathBuf;
|
||||
use std::time::Duration;
|
||||
|
||||
use anyhow::Context;
|
||||
use bsk_protocol::Method;
|
||||
use bsk_protocol::tools::{
|
||||
RecordAwaitParams, RecordAwaitResult, RecordStartParams, RecordStartResult, RecordStopParams,
|
||||
RecordStopResult, Trace,
|
||||
};
|
||||
use clap::{Args, Subcommand};
|
||||
|
||||
use crate::cli::TOOL_IPC_TIMEOUT;
|
||||
use crate::cli::business_rpc;
|
||||
use crate::cli::ensure_daemon::ensure_daemon;
|
||||
use crate::cli::error::{CliError, Format};
|
||||
use crate::cli::record_state;
|
||||
use crate::cli::session::{start_session, stop_session};
|
||||
|
||||
/// Max wait for the user to click 结束 in the browser (24 hours).
|
||||
const RECORD_AWAIT_TIMEOUT_MS: u32 = 86_400_000;
|
||||
|
||||
/// `bsk record …` subcommand tree.
|
||||
#[derive(Debug, Clone, Args)]
|
||||
pub struct RecordCmd {
|
||||
#[command(subcommand)]
|
||||
pub sub: RecordSub,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Subcommand)]
|
||||
pub enum RecordSub {
|
||||
/// Open the Agent Window, record user actions, and block until finished in the browser.
|
||||
Start(RecordStartArgs),
|
||||
/// Stop recording from the terminal (fallback), write trace JSON, and close the window.
|
||||
/// Works even while `record start` is blocked in `record_await` (daemon forwards
|
||||
/// `tool.record_stop` without the per-session busy gate).
|
||||
Stop(RecordStopArgs),
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Args)]
|
||||
pub struct RecordStartArgs {
|
||||
#[arg(long = "tab-id")]
|
||||
pub tab_id: Option<i64>,
|
||||
|
||||
/// Navigate to this http(s) URL before recording (required for a fresh
|
||||
/// session — Agent Window starts on about:blank, which cannot host the
|
||||
/// content script).
|
||||
#[arg(long)]
|
||||
pub url: Option<String>,
|
||||
|
||||
/// Optional goal text stored on the exported trace for LLM context.
|
||||
#[arg(long)]
|
||||
pub purpose: Option<String>,
|
||||
|
||||
/// Output path for the trace JSON (default `./trace.json`).
|
||||
#[arg(long, default_value = "trace.json")]
|
||||
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")]
|
||||
pub output: PathBuf,
|
||||
}
|
||||
|
||||
pub fn dispatch(cmd: RecordCmd, format: Format) -> Result<(), CliError> {
|
||||
match cmd.sub {
|
||||
RecordSub::Start(args) => dispatch_start(args, format),
|
||||
RecordSub::Stop(args) => dispatch_stop(args, format),
|
||||
}
|
||||
}
|
||||
|
||||
fn dispatch_start(args: RecordStartArgs, format: Format) -> Result<(), CliError> {
|
||||
if record_state::read().is_ok() {
|
||||
return Err(CliError::Local(anyhow::anyhow!(
|
||||
"a recording is already in progress; run `bsk record stop` first"
|
||||
)));
|
||||
}
|
||||
|
||||
let info = ensure_daemon().context("ensure daemon is running")?;
|
||||
let session = start_session(info.sock_path.clone(), None)?;
|
||||
|
||||
let start_params = RecordStartParams {
|
||||
session_id: session.session_id.clone(),
|
||||
tab_id: args.tab_id,
|
||||
url: args.url,
|
||||
purpose: args.purpose.clone(),
|
||||
};
|
||||
let start_result = business_rpc::call::<RecordStartParams, RecordStartResult>(
|
||||
info.sock_path.clone(),
|
||||
"record-start",
|
||||
Method::ToolRecordStart,
|
||||
Some(start_params),
|
||||
TOOL_IPC_TIMEOUT,
|
||||
);
|
||||
|
||||
let start_result = match start_result {
|
||||
Ok(result) => result,
|
||||
Err(err) => {
|
||||
let _ = stop_session(info.sock_path, &session.session_id);
|
||||
return Err(err);
|
||||
}
|
||||
};
|
||||
|
||||
if let Err(err) = record_state::write(&session.session_id) {
|
||||
let _ = stop_session(info.sock_path.clone(), &session.session_id);
|
||||
return Err(CliError::Local(err));
|
||||
}
|
||||
|
||||
if format == Format::Human {
|
||||
println!(
|
||||
"recording on tab={} — click 结束 in the browser when done",
|
||||
start_result.tab_id
|
||||
);
|
||||
}
|
||||
|
||||
let await_params = RecordAwaitParams {
|
||||
session_id: session.session_id.clone(),
|
||||
timeout_ms: Some(RECORD_AWAIT_TIMEOUT_MS),
|
||||
};
|
||||
let await_result = business_rpc::call::<RecordAwaitParams, RecordAwaitResult>(
|
||||
info.sock_path.clone(),
|
||||
"record-await",
|
||||
Method::ToolRecordAwait,
|
||||
Some(await_params),
|
||||
record_await_ipc_timeout(RECORD_AWAIT_TIMEOUT_MS),
|
||||
);
|
||||
|
||||
// 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)
|
||||
})(),
|
||||
Err(err) => Err(err),
|
||||
};
|
||||
|
||||
let session_stop_result = stop_session(info.sock_path, &session.session_id);
|
||||
record_state::clear();
|
||||
|
||||
run_result?;
|
||||
session_stop_result?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn dispatch_stop(args: RecordStopArgs, format: Format) -> Result<(), CliError> {
|
||||
let state = record_state::read().map_err(CliError::Local)?;
|
||||
let info = ensure_daemon().context("ensure daemon is running")?;
|
||||
let session_id = state.session_id.clone();
|
||||
|
||||
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 session_stop_result = stop_session(info.sock_path, &session_id);
|
||||
record_state::clear();
|
||||
|
||||
run_result?;
|
||||
session_stop_result?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn record_await_ipc_timeout(timeout_ms: u32) -> Duration {
|
||||
Duration::from_millis(u64::from(timeout_ms))
|
||||
.checked_add(Duration::from_secs(15))
|
||||
.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)?;
|
||||
}
|
||||
}
|
||||
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> {
|
||||
match format {
|
||||
Format::Json => {
|
||||
println!(
|
||||
"{}",
|
||||
serde_json::to_string(&serde_json::json!({
|
||||
"output": output,
|
||||
"trace": trace,
|
||||
"window_closed": true,
|
||||
}))
|
||||
.map_err(|e| CliError::Local(anyhow::anyhow!(e)))?
|
||||
);
|
||||
}
|
||||
Format::Human => {
|
||||
println!("saved {} steps to {}", trace.steps.len(), output.display());
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn render_stop(
|
||||
result: &RecordStopResult,
|
||||
output: &PathBuf,
|
||||
format: Format,
|
||||
) -> Result<(), CliError> {
|
||||
render_finish(&result.trace, output, format)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn default_output_is_trace_json() {
|
||||
let args = RecordStopArgs {
|
||||
output: PathBuf::from("trace.json"),
|
||||
};
|
||||
assert_eq!(args.output, PathBuf::from("trace.json"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn start_args_default_output_is_trace_json() {
|
||||
let args = RecordStartArgs {
|
||||
tab_id: None,
|
||||
url: None,
|
||||
purpose: None,
|
||||
output: PathBuf::from("trace.json"),
|
||||
};
|
||||
assert_eq!(args.output, PathBuf::from("trace.json"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn record_await_ipc_timeout_covers_long_wait() {
|
||||
let got = record_await_ipc_timeout(RECORD_AWAIT_TIMEOUT_MS);
|
||||
assert!(got >= Duration::from_secs(86_400));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,117 @@
|
||||
//! Persist the session id between `bsk record start` and `bsk record stop`.
|
||||
|
||||
use std::fs;
|
||||
use std::time::{SystemTime, UNIX_EPOCH};
|
||||
|
||||
use anyhow::{Context, Result};
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
use crate::daemon::paths;
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct RecordSessionState {
|
||||
pub session_id: String,
|
||||
pub started_at: String,
|
||||
}
|
||||
|
||||
pub fn write(session_id: &str) -> Result<()> {
|
||||
let path = paths::record_session_path()?;
|
||||
paths::ensure_bsk_home()?;
|
||||
if path.exists() {
|
||||
return Err(anyhow::anyhow!(
|
||||
"a recording is already in progress; run `bsk record stop` first"
|
||||
));
|
||||
}
|
||||
let state = RecordSessionState {
|
||||
session_id: session_id.to_string(),
|
||||
started_at: started_at_unix_ms(),
|
||||
};
|
||||
let json = serde_json::to_string_pretty(&state).context("serialize record session state")?;
|
||||
fs::write(&path, format!("{json}\n")).with_context(|| format!("write {}", path.display()))?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn read() -> Result<RecordSessionState> {
|
||||
let path = paths::record_session_path()?;
|
||||
if !path.exists() {
|
||||
return Err(anyhow::anyhow!(
|
||||
"no recording in progress; run `bsk record start` first"
|
||||
));
|
||||
}
|
||||
let raw = fs::read_to_string(&path).with_context(|| format!("read {}", path.display()))?;
|
||||
serde_json::from_str(&raw).with_context(|| format!("parse {}", path.display()))
|
||||
}
|
||||
|
||||
pub fn clear() {
|
||||
if let Ok(path) = paths::record_session_path() {
|
||||
let _ = fs::remove_file(path);
|
||||
}
|
||||
}
|
||||
|
||||
fn started_at_unix_ms() -> String {
|
||||
let ms = SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.unwrap_or_default()
|
||||
.as_millis();
|
||||
format!("{ms}")
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use std::sync::Mutex;
|
||||
|
||||
fn env_guard() -> &'static Mutex<()> {
|
||||
static GUARD: Mutex<()> = Mutex::new(());
|
||||
&GUARD
|
||||
}
|
||||
|
||||
fn with_temp_home<F: FnOnce()>(f: F) {
|
||||
let _lock = env_guard().lock().unwrap_or_else(|e| e.into_inner());
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
unsafe {
|
||||
std::env::set_var(crate::daemon::paths::BSK_HOME_ENV, tmp.path());
|
||||
}
|
||||
f();
|
||||
clear();
|
||||
unsafe {
|
||||
std::env::remove_var(crate::daemon::paths::BSK_HOME_ENV);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn write_read_round_trips() {
|
||||
with_temp_home(|| {
|
||||
write("abcd").unwrap();
|
||||
let state = read().unwrap();
|
||||
assert_eq!(state.session_id, "abcd");
|
||||
assert!(!state.started_at.is_empty());
|
||||
});
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn write_rejects_duplicate() {
|
||||
with_temp_home(|| {
|
||||
write("abcd").unwrap();
|
||||
let err = write("efgh").unwrap_err();
|
||||
assert!(err.to_string().contains("already in progress"));
|
||||
});
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn read_errors_when_missing() {
|
||||
with_temp_home(|| {
|
||||
let err = read().unwrap_err();
|
||||
assert!(err.to_string().contains("no recording in progress"));
|
||||
});
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn clear_removes_state_file() {
|
||||
with_temp_home(|| {
|
||||
write("abcd").unwrap();
|
||||
clear();
|
||||
assert!(read().is_err());
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -63,11 +63,11 @@ struct StartParams {
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
struct StartReply {
|
||||
session_id: String,
|
||||
browser_instance_id: String,
|
||||
pub struct StartReply {
|
||||
pub session_id: String,
|
||||
pub browser_instance_id: String,
|
||||
#[serde(default)]
|
||||
agent_window_id: Option<i64>,
|
||||
pub agent_window_id: Option<i64>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Serialize)]
|
||||
@@ -78,21 +78,21 @@ struct StopParams {
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
struct StopReply {
|
||||
stopped: Vec<String>,
|
||||
pub struct StopReply {
|
||||
pub stopped: Vec<String>,
|
||||
#[serde(default)]
|
||||
failed: Vec<StopFailure>,
|
||||
pub failed: Vec<StopFailure>,
|
||||
#[serde(default)]
|
||||
returned_tab_ids: Vec<i64>,
|
||||
pub returned_tab_ids: Vec<i64>,
|
||||
#[serde(default)]
|
||||
return_failures: Vec<ReturnFailure>,
|
||||
pub return_failures: Vec<ReturnFailure>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize, Serialize)]
|
||||
struct StopFailure {
|
||||
session_id: String,
|
||||
code: ErrorCode,
|
||||
message: String,
|
||||
pub struct StopFailure {
|
||||
pub session_id: String,
|
||||
pub code: ErrorCode,
|
||||
pub message: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
@@ -113,14 +113,7 @@ pub fn dispatch(cmd: SessionCmd, format: Format) -> Result<(), CliError> {
|
||||
}
|
||||
|
||||
fn run_start(sock: PathBuf, args: SessionStartArgs, format: Format) -> Result<(), CliError> {
|
||||
let result: Result<StartReply, CliError> = call(
|
||||
sock,
|
||||
Method::SessionStart,
|
||||
Some(StartParams {
|
||||
browser_instance_id: args.browser,
|
||||
}),
|
||||
Duration::from_secs(30),
|
||||
);
|
||||
let result = start_session(sock, args.browser);
|
||||
match result {
|
||||
Ok(reply) => match format {
|
||||
Format::Json => {
|
||||
@@ -143,6 +136,31 @@ fn run_start(sock: PathBuf, args: SessionStartArgs, format: Format) -> Result<()
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Start a session and open the Agent Window. Used by `session start` and `record start`.
|
||||
pub fn start_session(sock: PathBuf, browser: Option<String>) -> Result<StartReply, CliError> {
|
||||
call(
|
||||
sock,
|
||||
Method::SessionStart,
|
||||
Some(StartParams {
|
||||
browser_instance_id: browser,
|
||||
}),
|
||||
Duration::from_secs(30),
|
||||
)
|
||||
}
|
||||
|
||||
/// Stop a single session by id.
|
||||
pub fn stop_session(sock: PathBuf, session_id: &str) -> Result<StopReply, CliError> {
|
||||
call(
|
||||
sock,
|
||||
Method::SessionStop,
|
||||
Some(StopParams {
|
||||
session_id: Some(session_id.to_string()),
|
||||
all: false,
|
||||
}),
|
||||
SESSION_STOP_IPC_TIMEOUT,
|
||||
)
|
||||
}
|
||||
|
||||
/// Render a `session.start` failure (review I3).
|
||||
///
|
||||
/// The previous implementation hand-rolled `eprintln!("error: ...")`
|
||||
|
||||
@@ -242,7 +242,10 @@ pub fn full_handler(status: DaemonStatus, state: Arc<DaemonState>) -> RpcHandler
|
||||
| Method::ToolSelect
|
||||
| Method::ToolEvaluate
|
||||
| Method::ToolWaitForNavigation
|
||||
| Method::ToolRequestHelp => {
|
||||
| Method::ToolRequestHelp
|
||||
| Method::ToolRecordStart
|
||||
| Method::ToolRecordStop
|
||||
| Method::ToolRecordAwait => {
|
||||
handle_tool_dispatch(&state, rpc_id, method, params).await
|
||||
}
|
||||
Method::ToolWaitMs => handle_wait_ms(&state.abort_registry, rpc_id, params).await,
|
||||
@@ -319,10 +322,19 @@ async fn handle_tool_dispatch(
|
||||
}
|
||||
};
|
||||
let entry = inflight_guard.entry();
|
||||
let outcome = state
|
||||
.tool_queues
|
||||
.dispatch(&session_id, method, params, timeout, Some(entry))
|
||||
.await;
|
||||
// `record_stop` must reach the extension while `record_await` holds the
|
||||
// serial busy lock — finishing the recording unblocks await.
|
||||
let outcome = if method == Method::ToolRecordStop {
|
||||
state
|
||||
.tool_queues
|
||||
.dispatch_unlocked(&session_id, method, params, timeout, Some(entry))
|
||||
.await
|
||||
} else {
|
||||
state
|
||||
.tool_queues
|
||||
.dispatch(&session_id, method, params, timeout, Some(entry))
|
||||
.await
|
||||
};
|
||||
drop(inflight_guard);
|
||||
match outcome {
|
||||
Ok(v) => ResponseBody::Ok(v),
|
||||
|
||||
@@ -95,6 +95,11 @@ pub fn sock_path() -> Result<PathBuf> {
|
||||
Ok(bsk_home()?.join("run").join("daemon.sock"))
|
||||
}
|
||||
|
||||
/// Path to the in-progress recording session state (`record-session.json`).
|
||||
pub fn record_session_path() -> Result<PathBuf> {
|
||||
Ok(bsk_home()?.join("record-session.json"))
|
||||
}
|
||||
|
||||
/// Windows named-pipe name. Include the resolved `BSK_HOME` path in the
|
||||
/// token so test homes and custom installs do not share a predictable
|
||||
/// per-username pipe.
|
||||
@@ -163,6 +168,10 @@ mod tests {
|
||||
assert_eq!(log_path().unwrap(), home.join("daemon.log"));
|
||||
assert_eq!(update_check_path().unwrap(), home.join("update-check.json"));
|
||||
assert_eq!(sock_path().unwrap(), home.join("run").join("daemon.sock"));
|
||||
assert_eq!(
|
||||
record_session_path().unwrap(),
|
||||
home.join("record-session.json")
|
||||
);
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
@@ -221,6 +221,13 @@ impl ToolQueueRegistry {
|
||||
guard.get(sid).is_some_and(|entry| entry.accepting)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
fn force_busy_for_tests(&self, sid: &SessionId, busy: bool) {
|
||||
let guard = self.queues.lock().expect("tool queue registry poisoned");
|
||||
let entry = guard.get(sid).expect("queue must exist");
|
||||
entry.state.lock().expect("queue state poisoned").busy = busy;
|
||||
}
|
||||
|
||||
/// Spawn the worker task for a session id. Idempotent: spawning the
|
||||
/// same id twice replaces the previous queue (the previous worker
|
||||
/// closes when its sender drops).
|
||||
@@ -320,6 +327,42 @@ impl ToolQueueRegistry {
|
||||
outcome
|
||||
}
|
||||
|
||||
/// Forward a tool RPC to the extension without taking the
|
||||
/// per-session busy lock. Used by `tool.record_stop` so a second
|
||||
/// CLI can finish an in-flight `tool.record_await` (extension
|
||||
/// dispatch already runs concurrent `void dispatch(msg)` handlers).
|
||||
pub async fn dispatch_unlocked(
|
||||
&self,
|
||||
sid: &SessionId,
|
||||
method: Method,
|
||||
params: Value,
|
||||
timeout: Duration,
|
||||
inflight: Option<Arc<ToolInflightEntry>>,
|
||||
) -> Result<Value, DispatchError> {
|
||||
{
|
||||
let guard = self.queues.lock().expect("tool queue registry poisoned");
|
||||
let Some(entry) = guard.get(sid) else {
|
||||
return Err(DispatchError::SessionNotFound);
|
||||
};
|
||||
if !entry.accepting {
|
||||
return Err(DispatchError::SessionStopping);
|
||||
}
|
||||
}
|
||||
let (respond_tx, _respond_rx) = oneshot::channel();
|
||||
let job = ToolJob {
|
||||
method,
|
||||
params,
|
||||
timeout,
|
||||
respond: respond_tx,
|
||||
inflight,
|
||||
session_id: sid.clone(),
|
||||
};
|
||||
match forward_one(sid, &self.browsers, &self.sessions, &job).await {
|
||||
Ok(v) => Ok(v),
|
||||
Err(err) => Err(DispatchError::Rpc(err)),
|
||||
}
|
||||
}
|
||||
|
||||
/// Stop accepting new jobs for `sid`, enqueue one final control RPC,
|
||||
/// and wait for it behind any already-queued tools. Used by
|
||||
/// `session.stop` so extension teardown cannot race an in-flight
|
||||
@@ -789,3 +832,50 @@ mod tool_job_session_id_tests {
|
||||
assert_eq!(job.session_id.0, "sess-A");
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod dispatch_unlocked_tests {
|
||||
use super::*;
|
||||
use crate::daemon::browsers::BrowserRegistry;
|
||||
use crate::daemon::sessions::{SessionId, SessionRegistry};
|
||||
use bsk_protocol::Method;
|
||||
use serde_json::json;
|
||||
use std::sync::Arc;
|
||||
|
||||
#[tokio::test]
|
||||
async fn unlocked_dispatch_skips_session_busy_gate() {
|
||||
let browsers = Arc::new(BrowserRegistry::new());
|
||||
let sessions = Arc::new(SessionRegistry::new());
|
||||
let queues = ToolQueueRegistry::new(browsers, sessions);
|
||||
let sid = SessionId("s-record".into());
|
||||
queues.spawn(sid.clone());
|
||||
queues.force_busy_for_tests(&sid, true);
|
||||
|
||||
let busy = queues
|
||||
.dispatch(
|
||||
&sid,
|
||||
Method::ToolTabList,
|
||||
json!({ "session_id": "s-record" }),
|
||||
Duration::from_secs(1),
|
||||
None,
|
||||
)
|
||||
.await;
|
||||
assert!(matches!(busy, Err(DispatchError::SessionBusy)));
|
||||
|
||||
let unlocked = queues
|
||||
.dispatch_unlocked(
|
||||
&sid,
|
||||
Method::ToolRecordStop,
|
||||
json!({ "session_id": "s-record" }),
|
||||
Duration::from_secs(1),
|
||||
None,
|
||||
)
|
||||
.await;
|
||||
assert!(
|
||||
!matches!(unlocked, Err(DispatchError::SessionBusy)),
|
||||
"record_stop must not be rejected as SessionBusy; got {unlocked:?}"
|
||||
);
|
||||
// No session row → forward_one returns NotFound as Rpc.
|
||||
assert!(matches!(unlocked, Err(DispatchError::Rpc(_))));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -97,6 +97,7 @@ fn dispatch(cli: Cli, format: Format) -> Result<(), CliError> {
|
||||
Command::WaitForNavigation(args) => cli::waits::dispatch_wait_for_navigation(args, format),
|
||||
Command::WaitMs(args) => cli::waits::dispatch_wait_ms(args, format),
|
||||
Command::RequestHelp(args) => cli::human_loop::dispatch(args, format),
|
||||
Command::Record(cmd) => cli::record::dispatch(cmd, format),
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user