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

chore: add session-loss diagnostics

Browse Source
This commit is contained in:
drakezhang
2026-09-01 20:16:32 +08:00
parent f06a514bd2
commit b449be1544
19 changed files with 907 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/bsk-cli/src/daemon/browsers.rs
+25 -2
View File
@@ -223,6 +223,14 @@ impl BrowserRegistry {
"browser reconnect: replacing previous registration"
);
}
tracing::info!(
diagnostic = true,
event = "browser.registry.insert",
id = %client.id,
generation = client.generation,
previous_generation = guard.get(&client.id).map(|previous| previous.generation),
"diagnostic browser registry mutation"
);
guard.insert(client.id.clone(), client);
}
@@ -239,16 +247,31 @@ impl BrowserRegistry {
/// took over the same `id` while the old socket was tearing down
/// (review M4/M5 round 2 Important #1). Returns the removed entry
/// when a removal actually happened.
#[track_caller]
pub fn remove_if_generation_matches(
&self,
id: &BrowserId,
generation: u64,
) -> Option<std::sync::Arc<BrowserClient>> {
let caller = std::panic::Location::caller();
let mut guard = self.inner.lock().expect("browser registry poisoned");
match guard.get(id) {
let current_generation = guard.get(id).map(|current| current.generation);
let removed = match guard.get(id) {
Some(current) if current.generation == generation => guard.remove(id),
_ => None,
}
};
tracing::info!(
diagnostic = true,
event = "browser.registry.remove_if_generation_matches",
id = %id,
requested_generation = generation,
?current_generation,
removed = removed.is_some(),
caller_file = caller.file(),
caller_line = caller.line(),
"diagnostic browser registry mutation"
);
removed
}
pub fn get(&self, id: &BrowserId) -> Option<std::sync::Arc<BrowserClient>> {
crates/bsk-cli/src/daemon/ipc.rs
+48 -3
View File
@@ -37,7 +37,7 @@ use serde::{Deserialize, Serialize};
use serde_json::Value;
use tokio::sync::Notify;
use tokio::task::JoinHandle;
use tracing::{debug, warn};
use tracing::{debug, info, warn};
use super::abort::AbortRegistry;
use super::queue::{DEFAULT_TOOL_TIMEOUT, DispatchError};
@@ -197,7 +197,22 @@ pub fn full_handler(status: DaemonStatus, state: Arc<DaemonState>) -> RpcHandler
let status = status.clone();
let state = Arc::clone(&state);
Box::pin(async move {
match method {
let method_name = format!("{method:?}");
let diagnostic_rpc_id = rpc_id.clone();
let session_id = params
.get("session_id")
.and_then(Value::as_str)
.map(str::to_owned);
let started = Instant::now();
info!(
diagnostic = true,
event = "ipc.rpc.started",
rpc_id = %diagnostic_rpc_id,
method = %method_name,
session = session_id.as_deref().unwrap_or(""),
"diagnostic IPC RPC"
);
let body = match method {
Method::SystemPing => {
let result = PingResult { pong: true };
ResponseBody::Ok(serde_json::to_value(result).unwrap_or(Value::Null))
@@ -261,7 +276,23 @@ pub fn full_handler(status: DaemonStatus, state: Arc<DaemonState>) -> RpcHandler
message: format!("method not implemented yet: {other:?}"),
data: None,
}),
}
};
let (outcome, error_code) = match &body {
ResponseBody::Ok(_) => ("ok", String::new()),
ResponseBody::Err(error) => ("error", format!("{:?}", error.code)),
};
info!(
diagnostic = true,
event = "ipc.rpc.completed",
rpc_id = %diagnostic_rpc_id,
method = %method_name,
session = session_id.as_deref().unwrap_or(""),
outcome,
error_code = %error_code,
elapsed_ms = started.elapsed().as_millis() as u64,
"diagnostic IPC RPC"
);
body
})
})
}
@@ -465,8 +496,14 @@ fn handle_cancel(state: &Arc<DaemonState>, params: Value) -> ResponseBody {
};
let local = state.abort_registry.cancel(&params.rpc_id);
let mut cancelled = local;
let mut surface = if local { "abort_registry" } else { "none" };
if !local && let Some(snap) = state.tool_inflight.cancel(&params.rpc_id) {
cancelled = true;
surface = if snap.ws_rpc_id.is_some() {
"tool_inflight_forwarded"
} else {
"tool_inflight_queued"
};
// Forward the WS-side cancel only when the worker has
// already promoted the entry to "forwarded"; queued entries
// short-circuit on their own pre-flight without ever
@@ -484,6 +521,14 @@ fn handle_cancel(state: &Arc<DaemonState>, params: Value) -> ResponseBody {
);
}
}
info!(
diagnostic = true,
event = "ipc.cancel.resolved",
target_rpc_id = %params.rpc_id,
cancelled,
surface,
"diagnostic IPC cancellation"
);
ResponseBody::Ok(serde_json::to_value(CancelResult { cancelled }).unwrap_or(Value::Null))
}
crates/bsk-cli/src/daemon/queue.rs
+48 -1
View File
@@ -41,7 +41,7 @@ use bsk_protocol::{ErrorCode, Frame, Method, RequestFrame, ResponseBody, RpcErro
use rand::Rng;
use serde_json::Value;
use tokio::sync::{mpsc, oneshot};
use tracing::{debug, warn};
use tracing::{debug, info, warn};
use super::abort::AbortToken;
use super::browsers::BrowserRegistry;
@@ -515,6 +515,16 @@ async fn forward_one(
sessions: &Arc<SessionRegistry>,
job: &ToolJob,
) -> Result<Value, RpcError> {
let started = std::time::Instant::now();
info!(
diagnostic = true,
event = "tool.dispatch.started",
session = %sid,
method = ?job.method,
has_inflight = job.inflight.is_some(),
has_lifecycle_cancel = job.lifecycle_cancel.is_some(),
"diagnostic tool dispatch"
);
// Pre-flight: a cancel that landed while this job was still in
// the per-session channel must short-circuit before any session
// / browser resolution work, and before any WS frame leaves the
@@ -535,6 +545,13 @@ async fn forward_one(
));
}
let Some(session) = sessions.get(sid) else {
warn!(
diagnostic = true,
event = "tool.dispatch.session_missing",
session = %sid,
method = ?job.method,
"diagnostic tool dispatch"
);
return Err(RpcError {
code: ErrorCode::NotFound,
message: format!("session {sid} no longer exists"),
@@ -542,6 +559,14 @@ async fn forward_one(
});
};
let Some(client) = browsers.get(&session.browser_id) else {
warn!(
diagnostic = true,
event = "tool.dispatch.browser_missing",
session = %sid,
method = ?job.method,
browser = %session.browser_id,
"diagnostic tool dispatch"
);
return Err(RpcError {
code: ErrorCode::NotFound,
message: "owning browser is no longer connected".into(),
@@ -558,6 +583,16 @@ async fn forward_one(
method: job.method.clone(),
params: Some(job.params.clone()),
});
info!(
diagnostic = true,
event = "tool.dispatch.forwarding",
session = %sid,
method = ?job.method,
browser = %session.browser_id,
browser_generation = client.generation,
ws_rpc_id = %rpc_id,
"diagnostic tool dispatch"
);
// Promote the inflight entry to "forwarded" AND push the WS
// request frame to the sink inside the same critical section
// (review round 2 C1). The closure runs while the entry's inner
@@ -717,6 +752,18 @@ async fn forward_one(
});
}
};
info!(
diagnostic = true,
event = "tool.dispatch.response",
session = %sid,
method = ?job.method,
browser = %session.browser_id,
browser_generation = client.generation,
ws_rpc_id = %rpc_id,
elapsed_ms = started.elapsed().as_millis() as u64,
response_is_error = matches!(&response.body, ResponseBody::Err(_)),
"diagnostic tool dispatch"
);
match response.body {
ResponseBody::Ok(v) => Ok(v),
ResponseBody::Err(err) => Err(err),
crates/bsk-cli/src/daemon/sessions.rs
+68 -2
View File
@@ -105,12 +105,22 @@ impl SessionRegistry {
pub fn insert(&self, session: Session) {
let session_id = session.id.clone();
let browser_id = session.browser_id.clone();
let agent_window_id = session.agent_window_id;
let mut sessions = self.inner.lock().expect("session registry poisoned");
sessions.insert(session_id.clone(), session);
self.last_activity
.lock()
.expect("session activity registry poisoned")
.insert(session_id, Instant::now());
.insert(session_id.clone(), Instant::now());
tracing::info!(
diagnostic = true,
event = "session.registry.insert",
session = %session_id,
browser = %browser_id,
?agent_window_id,
"diagnostic session registry mutation"
);
}
/// Reserve a fresh, collision-free [`SessionId`] under the registry
@@ -152,6 +162,13 @@ impl SessionRegistry {
.lock()
.expect("session activity registry poisoned")
.insert(candidate.clone(), Instant::now());
tracing::info!(
diagnostic = true,
event = "session.registry.reserve",
session = %candidate,
browser = %browser_id,
"diagnostic session registry mutation"
);
return Some(candidate);
}
None
@@ -169,13 +186,24 @@ impl SessionRegistry {
let mut guard = self.inner.lock().expect("session registry poisoned");
let session = guard.get_mut(session_id)?;
session.agent_window_id = agent_window_id;
tracing::info!(
diagnostic = true,
event = "session.registry.commit",
session = %session_id,
browser = %session.browser_id,
?agent_window_id,
"diagnostic session registry mutation"
);
Some(session.clone())
}
/// Drop a placeholder reservation, used on extension error/timeout
/// paths so failed reservations do not accumulate.
#[track_caller]
pub fn cancel_reservation(&self, session_id: &SessionId) {
self.inner
let caller = std::panic::Location::caller();
let removed = self
.inner
.lock()
.expect("session registry poisoned")
.remove(session_id);
@@ -183,9 +211,20 @@ impl SessionRegistry {
.lock()
.expect("session activity registry poisoned")
.remove(session_id);
tracing::info!(
diagnostic = true,
event = "session.registry.cancel_reservation",
session = %session_id,
existed = removed.is_some(),
caller_file = caller.file(),
caller_line = caller.line(),
"diagnostic session registry mutation"
);
}
#[track_caller]
pub fn remove(&self, id: &SessionId) -> Option<Session> {
let caller = std::panic::Location::caller();
let removed = self
.inner
.lock()
@@ -195,6 +234,17 @@ impl SessionRegistry {
.lock()
.expect("session activity registry poisoned")
.remove(id);
tracing::info!(
diagnostic = true,
event = "session.registry.remove",
session = %id,
existed = removed.is_some(),
browser = removed.as_ref().map(|session| session.browser_id.0.as_str()),
agent_window_id = removed.as_ref().and_then(|session| session.agent_window_id),
caller_file = caller.file(),
caller_line = caller.line(),
"diagnostic session registry mutation"
);
removed
}
@@ -243,7 +293,9 @@ impl SessionRegistry {
}
/// Drop all sessions owned by `browser_id` (e.g. on disconnect).
#[track_caller]
pub fn purge_browser(&self, browser_id: &BrowserId) -> Vec<Session> {
let caller = std::panic::Location::caller();
let mut guard = self.inner.lock().expect("session registry poisoned");
let drained: Vec<Session> = guard
.values()
@@ -260,6 +312,20 @@ impl SessionRegistry {
for session in &drained {
activity.remove(&session.id);
}
let session_ids: Vec<&str> = drained
.iter()
.map(|session| session.id.0.as_str())
.collect();
tracing::info!(
diagnostic = true,
event = "session.registry.purge_browser",
browser = %browser_id,
sessions = ?session_ids,
count = drained.len(),
caller_file = caller.file(),
caller_line = caller.line(),
"diagnostic session registry mutation"
);
drained
}
crates/bsk-cli/src/daemon/start.rs
+11 -2
View File
@@ -40,6 +40,8 @@ pub(crate) const DAEMONIZED_ENV: &str = "BSK_DAEMONIZED";
/// daemon lock, IPC socket, and WS port — before taking over.
pub(crate) const DAEMON_REPLACEMENT_WAIT_ENV: &str = "BSK_DAEMON_REPLACES_PID";
const LUMI_DIAGNOSTIC_BUILD: &str = "lumi-session-loss-v1";
/// Concrete daemon configuration resolved from CLI flags / defaults.
#[derive(Debug, Clone)]
pub struct DaemonConfig {
@@ -251,6 +253,7 @@ pub fn run_foreground(cfg: DaemonConfig) -> Result<()> {
pid = info.pid,
ws_port = info.ws_port,
sock = %sock_path.display(),
diagnostic_build = LUMI_DIAGNOSTIC_BUILD,
"daemon ready"
);
@@ -588,7 +591,9 @@ pub(crate) fn spawn_update_check_task(
// The session gate is read after the fetch, as late
// as possible before the binary gets replaced.
let active_sessions = state.sessions.len();
let auto_update = update::auto_update_enabled() && exe_path.is_some();
// Keep this local diagnostic build in place until a
// reproduction has uploaded its extension-side evidence.
let auto_update = false;
update::auto_update_step(
candidate.as_ref(),
auto_update,
@@ -732,7 +737,11 @@ fn init_tracing() -> Option<tracing_appender::non_blocking::WorkerGuard> {
use tracing_subscriber::prelude::*;
let log_dir = paths::log_dir().ok();
let env_filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));
let env_filter = EnvFilter::try_from_default_env()
.unwrap_or_else(|_| EnvFilter::new("info"))
// Lumi commonly runs with RUST_LOG=warn. This local build must retain
// BrowserSkill's INFO diagnostics even under that host-wide setting.
.add_directive("bsk=info".parse().expect("valid diagnostic log directive"));
let (file_layer, guard) = if let Some(dir) = log_dir {
let appender = tracing_appender::rolling::daily(&dir, "daemon.log");
crates/bsk-cli/src/daemon/ws.rs
+65 -6
View File
@@ -233,6 +233,14 @@ async fn drive_connection(
.ok_or_else(|| anyhow!("handshake missing params"))?;
let params: HandshakeParams = serde_json::from_value(params_raw)
.map_err(|err| anyhow!("invalid HandshakeParams: {err}"))?;
info!(
diagnostic = true,
event = "browser.handshake.received",
browser = %params.instance_id,
extension_version = %params.version,
extension_protocol = %params.protocol_version,
"diagnostic browser handshake"
);
// Version compatibility (design §10, M10.4): protocol_version only.
let our_app_version: Version = env!("CARGO_PKG_VERSION")
@@ -350,7 +358,7 @@ async fn drive_connection(
// Pump loop: outbound frames from sink → ws; inbound from ws → resolve.
let pump_state = Arc::clone(&state);
let pump_browser = Arc::clone(&client);
let result_outcome: anyhow::Result<()> = async {
let result_outcome: anyhow::Result<String> = async {
loop {
tokio::select! {
outbound = rx.recv() => {
@@ -359,7 +367,7 @@ async fn drive_connection(
let json = serde_json::to_string(&frame)?;
writer.send(Message::Text(json)).await?;
}
None => break,
None => break Ok("outbound_channel_closed".to_string()),
}
}
msg = reader.next() => {
@@ -381,18 +389,37 @@ async fn drive_connection(
pump_browser.touch();
}
Some(Ok(Message::Frame(_))) => {}
Some(Ok(Message::Close(_))) | None => break,
Some(Ok(Message::Close(frame))) => {
let reason = frame
.map(|frame| format!("peer_close code={:?} reason={}", frame.code, frame.reason))
.unwrap_or_else(|| "peer_close_without_frame".to_string());
break Ok(reason);
}
None => break Ok("peer_eof".to_string()),
Some(Err(err)) => {
warn!(%err, "ws read error");
break;
break Ok(format!("read_error: {err}"));
}
}
}
}
}
Ok(())
}
.await;
let pump_exit_reason = match &result_outcome {
Ok(reason) => reason.clone(),
Err(err) => format!("pump_error: {err:#}"),
};
info!(
diagnostic = true,
event = "browser.websocket.pump_exit",
id = %browser_id,
generation,
reason = %pump_exit_reason,
last_seen_idle_ms = client.idle_for().as_millis() as u64,
session_ids = ?state.sessions.ids_for_browser(&browser_id),
"diagnostic browser websocket exit"
);
// Cleanup: drop browser + purge its sessions, but only if the
// registry still holds *this* generation. If a reconnect already
@@ -416,7 +443,7 @@ async fn drive_connection(
"browser disconnected; newer reconnect already owns this id, leaving sessions intact"
);
}
result_outcome
result_outcome.map(|_| ())
}
async fn handle_inbound_text(state: &Arc<DaemonState>, client: &Arc<BrowserClient>, text: &str) {
@@ -442,6 +469,38 @@ async fn handle_inbound_text(state: &Arc<DaemonState>, client: &Arc<BrowserClien
client.mark_heartbeat_seen();
debug!(id = %client.id, "heartbeat");
}
bsk_protocol::EventKind::BrowserDiagnostic => {
let entries = ev.payload.get("entries").and_then(|value| value.as_array());
info!(
diagnostic = true,
event = "extension.diagnostic.batch",
browser = %client.id,
generation = client.generation,
worker_boot_id = ev.payload.get("worker_boot_id").and_then(|value| value.as_str()).unwrap_or("unknown"),
count = entries.map_or(0, Vec::len),
"received extension diagnostic batch"
);
if let Some(entries) = entries {
for entry in entries {
info!(
diagnostic = true,
event = "extension.diagnostic.entry",
browser = %client.id,
generation = client.generation,
entry = %entry,
"extension diagnostic"
);
}
} else {
warn!(
diagnostic = true,
event = "extension.diagnostic.invalid",
browser = %client.id,
payload = %ev.payload,
"extension diagnostic payload missing entries"
);
}
}
bsk_protocol::EventKind::SessionWindowClosed => {
handle_session_window_closed(state, &client.id, &ev.payload);
}