From b449be1544406164afecc725fb7816093d79e825 Mon Sep 17 00:00:00 2001 From: drakezhang Date: Tue, 1 Sep 2026 20:13:59 +0800 Subject: [PATCH] chore: add session-loss diagnostics --- apps/extension/src/entrypoints/background.ts | 67 ++++++- .../src/lib/__tests__/diagnostics.test.ts | 61 +++++++ .../src/lib/connection-controller.ts | 94 +++++++++- apps/extension/src/lib/diagnostics.ts | 167 ++++++++++++++++++ apps/extension/src/lib/heartbeat.ts | 6 + apps/extension/src/lib/keepalive.ts | 19 +- .../src/session-manager/disconnect-cleanup.ts | 35 +++- .../src/session-manager/event-handler.ts | 16 ++ apps/extension/src/session-manager/manager.ts | 41 +++++ apps/extension/src/tools/dispatcher.ts | 67 +++++++ apps/extension/src/transport/ws-transport.ts | 55 +++++- apps/extension/wxt.config.ts | 5 +- crates/bsk-cli/src/daemon/browsers.rs | 27 ++- crates/bsk-cli/src/daemon/ipc.rs | 51 +++++- crates/bsk-cli/src/daemon/queue.rs | 49 ++++- crates/bsk-cli/src/daemon/sessions.rs | 70 +++++++- crates/bsk-cli/src/daemon/start.rs | 13 +- crates/bsk-cli/src/daemon/ws.rs | 71 +++++++- crates/bsk-protocol/src/frame.rs | 25 +++ 19 files changed, 907 insertions(+), 32 deletions(-) create mode 100644 apps/extension/src/lib/__tests__/diagnostics.test.ts create mode 100644 apps/extension/src/lib/diagnostics.ts diff --git a/apps/extension/src/entrypoints/background.ts b/apps/extension/src/entrypoints/background.ts index 4edd0a2..4d93d16 100644 --- a/apps/extension/src/entrypoints/background.ts +++ b/apps/extension/src/entrypoints/background.ts @@ -1,6 +1,7 @@ import { i18n } from "@browser-skill/i18n"; import { ChromiumCdp } from "@/browser-driver/chromium-cdp"; import { ConnectionController } from "@/lib/connection-controller"; +import { DiagnosticLogger } from "@/lib/diagnostics"; import { startHeartbeat } from "@/lib/heartbeat"; import { getConnectionEnabled, @@ -46,9 +47,20 @@ import type { Transport } from "@/transport/transport"; import { WSTransport } from "@/transport/ws-transport"; export default defineBackground(() => { - const controller = new ConnectionController(); - const transport = new WSTransport({ url: __BSK_DAEMON_WS_URL__ }); - const sessions = new SessionManager(); + const diagnostics = new DiagnosticLogger(); + diagnostics.record("extension.worker.started", { + diagnostic_build: "lumi-session-loss-v1", + extension_version: __BSK_EXT_VERSION__, + extension_runtime_id: chrome.runtime.id, + daemon_ws_url: __BSK_DAEMON_WS_URL__, + }); + const controller = new ConnectionController(diagnostics.sink); + const transport = new WSTransport({ + url: __BSK_DAEMON_WS_URL__, + diagnostics: diagnostics.sink, + }); + diagnostics.bindTransport((frame) => transport.send(frame)); + const sessions = new SessionManager({ diagnostics: diagnostics.sink }); const cdp = new ChromiumCdp(); const sessionsLive = attachSessionsLiveFlag({ manager: sessions }); let overlayGeneration = 0; @@ -109,6 +121,9 @@ export default defineBackground(() => { } function onOverlaySessionStateChanged(): void { + diagnostics.record("background.sessions.changed", { + live_session_ids: sessions.list().map((ctx) => ctx.sessionId), + }); void sessionsLive.syncFromManager(); const liveSessionIds = new Set(sessions.list().map((ctx) => ctx.sessionId)); for (const sessionId of controlModes.keys()) { @@ -146,6 +161,7 @@ export default defineBackground(() => { onSessionsChanged: () => { void sessionsLive.syncFromManager(); }, + diagnostics: diagnostics.sink, }); const recordDeps = { tabsApi: chrome.tabs, @@ -193,6 +209,7 @@ export default defineBackground(() => { title: i18n.t("helpRequest.notificationTitle", { ns: "extension" }), body: "", }), + diagnostics: diagnostics.sink, }); dispatcher.start(); if (typeof chrome.notifications?.onClicked?.addListener === "function") { @@ -225,6 +242,7 @@ export default defineBackground(() => { transport, cdp, onSessionsChanged: onOverlaySessionStateChanged, + diagnostics: diagnostics.sink, }); // MV3 service worker keepalive + reconnect supervisor (review M4/M5 @@ -236,6 +254,7 @@ export default defineBackground(() => { startKeepalive({ transport, shouldConnect: () => controller.isConnectionEnabled, + diagnostics: diagnostics.sink, }); // Application-level heartbeat (Chrome 116+): while the post-handshake @@ -249,6 +268,19 @@ export default defineBackground(() => { controller.subscribe((snap) => cb(snap.state === "connected" || snap.state === "version_skew"), ), + diagnostics: diagnostics.sink, + }); + + controller.subscribe((snapshot) => { + const ready = snapshot.state === "connected" || snapshot.state === "version_skew"; + diagnostics.setTransportReady(ready); + diagnostics.record("background.controller.snapshot", { + state: snapshot.state, + connection_enabled: snapshot.connectionEnabled, + instance_id: snapshot.instanceId, + last_error: snapshot.lastError, + live_session_ids: sessions.list().map((ctx) => ctx.sessionId), + }); }); // Wake-driven reconnect (best effort). An MV3 service worker is killed @@ -257,7 +289,12 @@ export default defineBackground(() => { // the worker and let us reconnect immediately instead of waiting for // the next 30s alarm tick. They only help when the daemon is actually // running; a cold daemon is (re)spawned by the next `bsk` command. - const reconnectIfNeeded = () => { + const reconnectIfNeeded = (source: string) => { + diagnostics.record("background.wake_reconnect.requested", { + source, + controller_enabled: controller.isConnectionEnabled, + transport_state: transport.state, + }); if (!controller.isConnectionEnabled) return; if (transport.state === "connected") return; void transport.connect().catch((err) => { @@ -265,13 +302,14 @@ export default defineBackground(() => { }); }; if (typeof chrome.runtime?.onStartup?.addListener === "function") { - chrome.runtime.onStartup.addListener(reconnectIfNeeded); + chrome.runtime.onStartup.addListener(() => reconnectIfNeeded("runtime.onStartup")); } if (typeof chrome.idle?.onStateChanged?.addListener === "function") { chrome.idle.onStateChanged.addListener((state) => { + diagnostics.record("chrome.idle.state_changed", { state }); // "active" fires when the user returns from idle/locked — the most // reliable "machine just woke" signal we get. - if (state === "active") reconnectIfNeeded(); + if (state === "active") reconnectIfNeeded("idle.active"); }); // Treat >60s of no input as idle so the active transition is timely. chrome.idle.setDetectionInterval?.(60); @@ -279,14 +317,21 @@ export default defineBackground(() => { void (async () => { const connectionEnabled = await getConnectionEnabled(); + diagnostics.record("background.controller.attach_started", { connectionEnabled }); await controller.attach(transport, detectBrowserMeta(), connectionEnabled, { beforeDisconnect: async () => { + diagnostics.record("background.before_disconnect_cleanup", { + live_session_ids: sessions.list().map((ctx) => ctx.sessionId), + }); const report = await cleanupAfterDisconnect(); if (report.failures.length > 0) { console.warn("[browser-skill] session cleanup before disconnect was incomplete", report); } }, onDisconnected: async () => { + diagnostics.record("background.after_disconnect_cleanup", { + live_session_ids: sessions.list().map((ctx) => ctx.sessionId), + }); const report = await cleanupAfterDisconnect(); if (report.failures.length > 0) { console.warn("[browser-skill] session cleanup after disconnect was incomplete", report); @@ -294,6 +339,7 @@ export default defineBackground(() => { }, }); })().catch((err) => { + diagnostics.record("background.controller.attach_failed", { error: err }); console.error("[browser-skill] controller failed to attach", err); }); @@ -374,6 +420,15 @@ export default defineBackground(() => { dbg.__bhDispatcher = dispatcher; } + if (typeof chrome.runtime?.onSuspend?.addListener === "function") { + chrome.runtime.onSuspend.addListener(() => { + diagnostics.record("extension.worker.suspending", { + live_session_ids: sessions.list().map((ctx) => ctx.sessionId), + transport_state: transport.state, + }); + }); + } + console.info("[browser-skill] background worker initialised"); }); diff --git a/apps/extension/src/lib/__tests__/diagnostics.test.ts b/apps/extension/src/lib/__tests__/diagnostics.test.ts new file mode 100644 index 0000000..ff9d39c --- /dev/null +++ b/apps/extension/src/lib/__tests__/diagnostics.test.ts @@ -0,0 +1,61 @@ +import { describe, expect, it, vi } from "vitest"; +import { DiagnosticLogger } from "../diagnostics"; + +class MemoryStorage { + value: Record = {}; + + async get(key: string): Promise> { + return { [key]: this.value[key] }; + } + + async set(items: Record): Promise { + this.value = { ...this.value, ...items }; + } +} + +describe("DiagnosticLogger", () => { + it("persists evidence before transport readiness and flushes it after reconnect", async () => { + const storage = new MemoryStorage(); + const send = vi.fn(); + const logger = new DiagnosticLogger(storage); + logger.bindTransport(send); + + logger.record("transport.socket.closed", { code: 1006 }); + await logger.flush(); + expect(send).not.toHaveBeenCalled(); + + logger.setTransportReady(true); + await logger.flush(); + + expect(send).toHaveBeenCalledOnce(); + expect(send.mock.calls[0]?.[0]).toMatchObject({ + event: "browser.diagnostic", + payload: { + schema_version: 1, + entries: [ + { + event: "transport.socket.closed", + fields: { code: 1006 }, + }, + ], + }, + }); + expect(storage.value.bsk_diagnostic_events_v1).toEqual([]); + }); + + it("keeps persisted evidence when the transport rejects the flush", async () => { + const storage = new MemoryStorage(); + const logger = new DiagnosticLogger(storage); + logger.bindTransport(() => { + throw new Error("socket closed"); + }); + logger.record("session.stop.requested", { session_id: "abcd" }); + + logger.setTransportReady(true); + await logger.flush(); + + expect(storage.value.bsk_diagnostic_events_v1).toMatchObject([ + { event: "session.stop.requested", fields: { session_id: "abcd" } }, + ]); + }); +}); diff --git a/apps/extension/src/lib/connection-controller.ts b/apps/extension/src/lib/connection-controller.ts index 687fe8a..caf7e6a 100644 --- a/apps/extension/src/lib/connection-controller.ts +++ b/apps/extension/src/lib/connection-controller.ts @@ -6,6 +6,7 @@ import { } from "../transport/handshake"; import type { Transport } from "../transport/transport"; import type { ConnectionState, HandshakeResult } from "../transport/types"; +import type { DiagnosticSink } from "./diagnostics"; import { getLabel, getOrCreateInstanceId } from "./instance-id"; import { compareProtocol, parseProtocolMajor } from "./semver"; @@ -51,6 +52,11 @@ export class ConnectionController { private handshakeAbort: AbortController | null = null; private handshakeRetryTimer: ReturnType | null = null; private suppressDisconnectRecovery = false; + private readonly diagnostics?: DiagnosticSink; + + constructor(diagnostics?: DiagnosticSink) { + this.diagnostics = diagnostics; + } get isConnectionEnabled(): boolean { return this.connectionEnabled; @@ -89,8 +95,20 @@ export class ConnectionController { this.lifecycleHooks = lifecycleHooks; this.instanceId = await getOrCreateInstanceId(); this.label = await getLabel(); + this.diagnostics?.("controller.attach", { + instance_id: this.instanceId, + connection_enabled: connectionEnabled, + transport_state: transport.state, + }); transport.onConnectionStateChange((s) => { + this.diagnostics?.("controller.transport_state", { + transport_state: s, + controller_state: this.currentState, + connection_generation: this.connectionGeneration, + suppress_disconnect_recovery: this.suppressDisconnectRecovery, + recovery_inflight: this.disconnectRecovery !== null, + }); if (s === "disconnected") { // An in-flight handshake belongs to the dead connection — cancel it so // a late resolution can never clobber the next attempt. @@ -103,6 +121,9 @@ export class ConnectionController { // handshake bounce); the caller owns the reconnect policy there. this.suppressDisconnectRecovery = false; } else { + this.diagnostics?.("controller.disconnect_recovery.requested", { + connection_generation: this.connectionGeneration, + }); void this.recoverFromDisconnect(); } } @@ -161,6 +182,12 @@ export class ConnectionController { const generation = ++this.connectionGeneration; const abort = new AbortController(); this.handshakeAbort = abort; + this.diagnostics?.("controller.handshake.started", { + connection_generation: generation, + browser_name: browser.name, + browser_version: browser.version, + instance_id: this.instanceId, + }); void this.runHandshake(browser, generation, abort.signal); } @@ -197,11 +224,21 @@ export class ConnectionController { } this.clearHandshakeRetry(); this.lastError = null; + this.diagnostics?.("controller.handshake.completed", { + connection_generation: generation, + verdict: verdict.kind, + daemon_version: outcome.result.version, + daemon_protocol: outcome.result.protocol_version, + }); this.setState(verdict.kind); } catch (err) { if (generation !== this.connectionGeneration || signal.aborted || isAbortError(err)) return; this.handshake = null; this.lastError = err instanceof Error ? err.message : String(err); + this.diagnostics?.("controller.handshake.failed", { + connection_generation: generation, + error: err, + }); // Bounce the half-open socket deliberately; the retry timer below owns // the reconnect, so skip unexpected-loss recovery for this disconnect. this.suppressDisconnectRecovery = true; @@ -228,23 +265,57 @@ export class ConnectionController { } private recoverFromDisconnect(): Promise { - if (this.disconnectRecovery) return this.disconnectRecovery; + if (this.disconnectRecovery) { + this.diagnostics?.("controller.disconnect_recovery.coalesced", { + connection_generation: this.connectionGeneration, + }); + return this.disconnectRecovery; + } const recovery = (async () => { + const recoveryGeneration = this.connectionGeneration; + this.diagnostics?.("controller.disconnect_recovery.started", { + recovery_generation: recoveryGeneration, + }); // WSTransport schedules its reconnect timer immediately after notifying // state listeners. Yield once, then disconnect explicitly so that timer // is cancelled before local session teardown begins. await Promise.resolve(); - if (!this.connectionEnabled || !this.transport) return; + if (!this.connectionEnabled || !this.transport) { + this.diagnostics?.("controller.disconnect_recovery.skipped", { + recovery_generation: recoveryGeneration, + reason: "disabled_or_missing_transport", + }); + return; + } // Another path already reconnected while we yielded — nothing to do. - if (this.currentState !== "disconnected") return; + if (this.currentState !== "disconnected") { + this.diagnostics?.("controller.disconnect_recovery.skipped", { + recovery_generation: recoveryGeneration, + reason: "state_already_changed", + current_state: this.currentState, + }); + return; + } await this.transport.disconnect().catch(() => {}); try { + this.diagnostics?.("controller.disconnect_cleanup.started", { + recovery_generation: recoveryGeneration, + current_generation: this.connectionGeneration, + }); await this.lifecycleHooks.onDisconnected?.(); + this.diagnostics?.("controller.disconnect_cleanup.completed", { + recovery_generation: recoveryGeneration, + current_generation: this.connectionGeneration, + }); } catch (err) { console.warn("[browser-skill] session cleanup after disconnect failed", err); } if (!this.connectionEnabled) return; try { + this.diagnostics?.("controller.disconnect_reconnect.started", { + recovery_generation: recoveryGeneration, + current_generation: this.connectionGeneration, + }); await this.transport.connect(); } catch (err) { this.lastError = err instanceof Error ? err.message : String(err); @@ -252,22 +323,39 @@ export class ConnectionController { } })(); this.disconnectRecovery = recovery.finally(() => { + this.diagnostics?.("controller.disconnect_recovery.completed", { + connection_generation: this.connectionGeneration, + current_state: this.currentState, + }); this.disconnectRecovery = null; }); return this.disconnectRecovery; } private cancelHandshake(): void { + const previousGeneration = this.connectionGeneration; this.connectionGeneration += 1; + this.diagnostics?.("controller.handshake.cancelled", { + previous_generation: previousGeneration, + next_generation: this.connectionGeneration, + had_abort_controller: this.handshakeAbort !== null, + }); this.handshakeAbort?.abort(); this.handshakeAbort = null; } private scheduleHandshakeRetry(browser: { name: string; version: string }): void { if (this.handshakeRetryTimer || !this.connectionEnabled) return; + this.diagnostics?.("controller.handshake_retry.scheduled", { + delay_ms: HANDSHAKE_RETRY_DELAY_MS, + connection_generation: this.connectionGeneration, + }); this.handshakeRetryTimer = setTimeout(() => { this.handshakeRetryTimer = null; if (!this.connectionEnabled || !this.transport) return; + this.diagnostics?.("controller.handshake_retry.timer_fired", { + connection_generation: this.connectionGeneration, + }); void this.transport.connect().catch((err) => { this.lastError = err instanceof Error ? err.message : String(err); this.setState("disconnected"); diff --git a/apps/extension/src/lib/diagnostics.ts b/apps/extension/src/lib/diagnostics.ts new file mode 100644 index 0000000..10fede8 --- /dev/null +++ b/apps/extension/src/lib/diagnostics.ts @@ -0,0 +1,167 @@ +import type { ProtocolFrame } from "@/transport/types"; + +const STORAGE_KEY = "bsk_diagnostic_events_v1"; +const MAX_ENTRIES = 500; + +export type DiagnosticFields = Record; +export type DiagnosticSink = (event: string, fields?: DiagnosticFields) => void; + +export interface DiagnosticEntry { + id: string; + sequence: number; + timestamp: string; + monotonic_ms: number; + worker_boot_id: string; + event: string; + fields: DiagnosticFields; +} + +interface DiagnosticStorage { + get(key: string): Promise>; + set(items: Record): Promise; +} + +function defaultStorage(): DiagnosticStorage | null { + if (typeof chrome === "undefined" || !chrome.storage?.local) return null; + return { + get: (key) => chrome.storage.local.get(key), + set: (items) => chrome.storage.local.set(items), + }; +} + +function makeBootId(): string { + try { + return crypto.randomUUID(); + } catch { + return `worker-${Date.now()}-${Math.random().toString(16).slice(2)}`; + } +} + +function normalise(value: unknown): unknown { + if (value instanceof Error) { + return { name: value.name, message: value.message, stack: value.stack }; + } + if (typeof value === "bigint") return value.toString(); + if (Array.isArray(value)) return value.map(normalise); + if (value && typeof value === "object") { + return Object.fromEntries( + Object.entries(value as Record).map(([key, child]) => [ + key, + normalise(child), + ]), + ); + } + return value; +} + +/** + * Persistent diagnostics for the MV3 background worker. + * + * Events are first written to chrome.storage.local so a worker termination or + * dead WebSocket does not erase the evidence. Once the post-handshake link is + * ready, pending entries are forwarded to the daemon as one + * `browser.diagnostic` event and removed from the local queue. + */ +export class DiagnosticLogger { + readonly workerBootId: string; + readonly sink: DiagnosticSink; + + private sequence = 0; + private storageQueue: Promise = Promise.resolve(); + private flushPromise: Promise | null = null; + private flushRequested = false; + private send: ((frame: ProtocolFrame) => void) | null = null; + private transportReady = false; + private readonly storage: DiagnosticStorage | null; + + constructor(storage: DiagnosticStorage | null = defaultStorage()) { + this.workerBootId = makeBootId(); + this.storage = storage; + this.sink = (event, fields = {}) => this.record(event, fields); + } + + bindTransport(send: (frame: ProtocolFrame) => void): void { + this.send = send; + } + + setTransportReady(ready: boolean): void { + this.transportReady = ready; + if (ready) void this.flush(); + } + + record(event: string, fields: DiagnosticFields = {}): void { + const sequence = ++this.sequence; + const entry: DiagnosticEntry = { + id: `${this.workerBootId}:${sequence}`, + sequence, + timestamp: new Date().toISOString(), + monotonic_ms: Math.round(performance.now()), + worker_boot_id: this.workerBootId, + event, + fields: normalise(fields) as DiagnosticFields, + }; + console.info("[bsk diagnostic]", entry); + this.storageQueue = this.storageQueue + .then(async () => { + if (!this.storage) return; + const stored = await this.storage.get(STORAGE_KEY); + const previous = Array.isArray(stored[STORAGE_KEY]) + ? (stored[STORAGE_KEY] as DiagnosticEntry[]) + : []; + await this.storage.set({ + [STORAGE_KEY]: [...previous, entry].slice(-MAX_ENTRIES), + }); + }) + .catch((error) => { + console.warn("[bsk diagnostic] failed to persist event", error); + }); + if (this.transportReady) void this.flush(); + } + + flush(): Promise { + if (this.flushPromise) { + this.flushRequested = true; + return this.flushPromise; + } + const running = (async () => { + await this.storageQueue; + if (!this.transportReady || !this.send || !this.storage) return; + const stored = await this.storage.get(STORAGE_KEY); + const entries = Array.isArray(stored[STORAGE_KEY]) + ? (stored[STORAGE_KEY] as DiagnosticEntry[]) + : []; + if (entries.length === 0) return; + try { + this.send({ + event: "browser.diagnostic", + payload: { + schema_version: 1, + worker_boot_id: this.workerBootId, + entries, + }, + }); + } catch (error) { + console.warn("[bsk diagnostic] daemon flush failed", error); + return; + } + const sentIds = new Set(entries.map((entry) => entry.id)); + const current = await this.storage.get(STORAGE_KEY); + const currentEntries = Array.isArray(current[STORAGE_KEY]) + ? (current[STORAGE_KEY] as DiagnosticEntry[]) + : []; + await this.storage.set({ + [STORAGE_KEY]: currentEntries.filter((entry) => !sentIds.has(entry.id)), + }); + })().catch((error) => { + console.warn("[bsk diagnostic] flush failed", error); + }); + this.flushPromise = running.finally(() => { + this.flushPromise = null; + if (this.flushRequested && this.transportReady) { + this.flushRequested = false; + void this.flush(); + } + }); + return this.flushPromise; + } +} diff --git a/apps/extension/src/lib/heartbeat.ts b/apps/extension/src/lib/heartbeat.ts index 7869291..2509c4f 100644 --- a/apps/extension/src/lib/heartbeat.ts +++ b/apps/extension/src/lib/heartbeat.ts @@ -21,6 +21,7 @@ * after the handshake has landed. */ +import type { DiagnosticSink } from "@/lib/diagnostics"; import type { ProtocolFrame } from "@/transport/types"; export const HEARTBEAT_EVENT = "system.heartbeat"; @@ -46,6 +47,7 @@ export interface HeartbeatOptions { intervalMs?: number; /** Inject fake timers (tests). */ timers?: HeartbeatTimers; + diagnostics?: DiagnosticSink; } export interface HeartbeatHandle { @@ -74,7 +76,9 @@ export function startHeartbeat(options: HeartbeatOptions): HeartbeatHandle { const beat = () => { try { options.send({ event: HEARTBEAT_EVENT, payload: {} }); + options.diagnostics?.("heartbeat.sent"); } catch (err) { + options.diagnostics?.("heartbeat.send_failed", { error: err }); // The socket raced a close between the state change and this // tick. The transport's own reconnect path recovers; swallow so // the interval keeps running (it is cleared on the disconnect @@ -87,12 +91,14 @@ export function startHeartbeat(options: HeartbeatOptions): HeartbeatHandle { if (timer !== null) { timers.clearInterval(timer); timer = null; + options.diagnostics?.("heartbeat.timer.stopped"); } }; const start = () => { if (timer !== null) return; timer = timers.setInterval(beat, intervalMs); + options.diagnostics?.("heartbeat.timer.started", { interval_ms: intervalMs }); }; const unsubscribe = options.onActiveChange((active) => { diff --git a/apps/extension/src/lib/keepalive.ts b/apps/extension/src/lib/keepalive.ts index f667c60..09d6b03 100644 --- a/apps/extension/src/lib/keepalive.ts +++ b/apps/extension/src/lib/keepalive.ts @@ -20,6 +20,7 @@ * SW idle threshold. */ +import type { DiagnosticSink } from "@/lib/diagnostics"; import type { Transport } from "@/transport/transport"; export const KEEPALIVE_ALARM_NAME = "bh-keepalive"; @@ -59,6 +60,7 @@ export interface KeepaliveOptions { alarmName?: string; /** Override the alarm period (tests). */ periodMin?: number; + diagnostics?: DiagnosticSink; } export interface KeepaliveHandle { @@ -79,11 +81,24 @@ export function startKeepalive(options: KeepaliveOptions): KeepaliveHandle { const period = options.periodMin ?? KEEPALIVE_PERIOD_MIN; const tick = async () => { - if (options.shouldConnect && !options.shouldConnect()) return; - if (options.transport.state === "connected") return; + if (options.shouldConnect && !options.shouldConnect()) { + options.diagnostics?.("keepalive.tick.skipped", { reason: "connection_disabled" }); + return; + } + if (options.transport.state === "connected") { + options.diagnostics?.("keepalive.tick.healthy", { transport_state: "connected" }); + return; + } + options.diagnostics?.("keepalive.reconnect.started", { + transport_state: options.transport.state, + }); try { await options.transport.connect(); + options.diagnostics?.("keepalive.reconnect.completed", { + transport_state: options.transport.state, + }); } catch (err) { + options.diagnostics?.("keepalive.reconnect.failed", { error: err }); console.debug("[bsk keepalive] connect attempt failed", err); } }; diff --git a/apps/extension/src/session-manager/disconnect-cleanup.ts b/apps/extension/src/session-manager/disconnect-cleanup.ts index 59daafb..3cbbc8f 100644 --- a/apps/extension/src/session-manager/disconnect-cleanup.ts +++ b/apps/extension/src/session-manager/disconnect-cleanup.ts @@ -1,3 +1,4 @@ +import type { DiagnosticSink } from "@/lib/diagnostics"; import { handleSessionStop, type SessionStopDeps } from "@/tools/session"; import type { SessionManager } from "./manager"; @@ -15,6 +16,7 @@ export interface DisconnectCleanupOptions { manager: SessionManager; sessionStopDeps?: SessionStopDeps; onSessionsChanged?: () => void; + diagnostics?: DiagnosticSink; } /** @@ -29,7 +31,15 @@ export function createDisconnectCleanup(options: DisconnectCleanupOptions) { let inFlight: Promise | null = null; return (): Promise => { - if (inFlight) return inFlight; + if (inFlight) { + options.diagnostics?.("disconnect_cleanup.coalesced", { + live_session_ids: options.manager.list().map((ctx) => ctx.sessionId), + }); + return inFlight; + } + options.diagnostics?.("disconnect_cleanup.scheduled", { + live_session_ids: options.manager.list().map((ctx) => ctx.sessionId), + }); inFlight = cleanupSessions(options).finally(() => { inFlight = null; }); @@ -42,8 +52,17 @@ async function cleanupSessions( ): Promise { const stoppedSessionIds: string[] = []; const failures: DisconnectCleanupFailure[] = []; + const capturedSessions = options.manager.list(); + options.diagnostics?.("disconnect_cleanup.started", { + captured_session_ids: capturedSessions.map((ctx) => ctx.sessionId), + }); - for (const ctx of options.manager.list()) { + for (const ctx of capturedSessions) { + options.diagnostics?.("disconnect_cleanup.session_stop.started", { + session_id: ctx.sessionId, + agent_window_id: ctx.agentWindowId, + still_registered: options.manager.has(ctx.sessionId), + }); try { const result = await handleSessionStop( options.manager, @@ -62,14 +81,26 @@ async function cleanupSessions( continue; } stoppedSessionIds.push(ctx.sessionId); + options.diagnostics?.("disconnect_cleanup.session_stop.completed", { + session_id: ctx.sessionId, + }); } catch (err) { failures.push({ sessionId: ctx.sessionId, message: err instanceof Error ? err.message : String(err), }); + options.diagnostics?.("disconnect_cleanup.session_stop.failed", { + session_id: ctx.sessionId, + error: err, + }); } } options.onSessionsChanged?.(); + options.diagnostics?.("disconnect_cleanup.completed", { + stopped_session_ids: stoppedSessionIds, + failures, + remaining_session_ids: options.manager.list().map((ctx) => ctx.sessionId), + }); return { stoppedSessionIds, failures }; } diff --git a/apps/extension/src/session-manager/event-handler.ts b/apps/extension/src/session-manager/event-handler.ts index 5afe53c..07aa879 100644 --- a/apps/extension/src/session-manager/event-handler.ts +++ b/apps/extension/src/session-manager/event-handler.ts @@ -1,3 +1,4 @@ +import type { DiagnosticSink } from "@/lib/diagnostics"; import type { Transport } from "@/transport/transport"; import type { EventFrame } from "@/transport/types"; import type { SessionManager } from "./manager"; @@ -24,6 +25,7 @@ export interface SessionEventHandlerOptions { * `chrome.storage.session` "sessions live" flag (review M4/M5 I3). */ onSessionsChanged?: () => void; + diagnostics?: DiagnosticSink; } function chromeWindowEvents(): WindowRemovedListener { @@ -50,6 +52,11 @@ export function attachSessionEventHandler(options: SessionEventHandlerOptions): const onRemoved = (windowId: number): void => { const ctx = manager.findByWindowId(windowId); + options.diagnostics?.("chrome.window.removed", { + window_id: windowId, + matched_session_id: ctx?.sessionId ?? null, + live_session_ids: manager.list().map((entry) => entry.sessionId), + }); if (!ctx) return; const returnFailures = Array.from(ctx.borrowedTabs.keys()).map((tabId) => ({ tab_id: tabId, @@ -81,7 +88,16 @@ export function attachSessionEventHandler(options: SessionEventHandlerOptions): }; try { transport.send(event); + options.diagnostics?.("session.window_closed.sent", { + session_id: ctx.sessionId, + window_id: windowId, + }); } catch (err) { + options.diagnostics?.("session.window_closed.send_failed", { + session_id: ctx.sessionId, + window_id: windowId, + error: err, + }); console.warn("[bh] could not push session.window_closed event", err); } }) diff --git a/apps/extension/src/session-manager/manager.ts b/apps/extension/src/session-manager/manager.ts index 2fef0d7..b9075ba 100644 --- a/apps/extension/src/session-manager/manager.ts +++ b/apps/extension/src/session-manager/manager.ts @@ -1,3 +1,4 @@ +import type { DiagnosticSink } from "@/lib/diagnostics"; import { AGENT_WINDOW_HOME, type AgentWindowApi, chromeAgentWindowApi } from "./agent-window"; import { RefStore } from "./ref-store"; @@ -23,6 +24,7 @@ export interface BorrowReservation { export interface SessionManagerOptions { agentWindow?: AgentWindowApi; now?: () => number; + diagnostics?: DiagnosticSink; } /** Options for starting a session's Agent Window. */ @@ -83,10 +85,12 @@ export class SessionManager { private readonly borrowReservations = new Map(); private readonly agentWindow: AgentWindowApi; private readonly now: () => number; + private readonly diagnostics?: DiagnosticSink; constructor(options: SessionManagerOptions = {}) { this.agentWindow = options.agentWindow ?? chromeAgentWindowApi; this.now = options.now ?? Date.now; + this.diagnostics = options.diagnostics; } has(sessionId: string): boolean { @@ -169,6 +173,13 @@ export class SessionManager { * daemon in the `tool.session_start` reply. */ async start(sessionId: string, opts: SessionStartOptions = {}): Promise { + this.diagnostics?.("session_manager.start.requested", { + session_id: sessionId, + existing_session_ids: this.list().map((ctx) => ctx.sessionId), + focused: opts.focused ?? null, + size: opts.size ?? null, + signal_aborted: opts.signal?.aborted ?? false, + }); if (this.sessions.has(sessionId)) { throw new Error(`[bh] session ${sessionId} already exists`); } @@ -178,6 +189,10 @@ export class SessionManager { try { const { signal: _signal, ...createOptions } = opts; windowId = await this.agentWindow.create(AGENT_WINDOW_HOME, createOptions); + this.diagnostics?.("session_manager.window.created", { + session_id: sessionId, + agent_window_id: windowId, + }); throwIfSessionStartAborted(opts.signal); await this.agentWindow.ensureActiveTab(windowId, AGENT_WINDOW_HOME); throwIfSessionStartAborted(opts.signal); @@ -191,8 +206,18 @@ export class SessionManager { }; this.sessions.set(sessionId, ctx); this.windowIndex.set(windowId, sessionId); + this.diagnostics?.("session_manager.start.committed", { + session_id: sessionId, + agent_window_id: windowId, + live_session_ids: this.list().map((entry) => entry.sessionId), + }); return ctx; } catch (startupError) { + this.diagnostics?.("session_manager.start.failed", { + session_id: sessionId, + agent_window_id: windowId, + error: startupError, + }); if (windowId !== null) { try { await this.agentWindow.remove(windowId); @@ -216,12 +241,28 @@ export class SessionManager { options: { dropOnly?: boolean } = {}, ): Promise { const ctx = this.sessions.get(sessionId); + this.diagnostics?.("session_manager.stop.requested", { + session_id: sessionId, + drop_only: options.dropOnly ?? false, + existed: ctx !== undefined, + live_session_ids: this.list().map((entry) => entry.sessionId), + }); if (!ctx) return null; if (!options.dropOnly) { await this.agentWindow.remove(ctx.agentWindowId); + this.diagnostics?.("session_manager.window.removed", { + session_id: sessionId, + agent_window_id: ctx.agentWindowId, + }); } this.sessions.delete(sessionId); this.windowIndex.delete(ctx.agentWindowId); + this.diagnostics?.("session_manager.stop.committed", { + session_id: sessionId, + agent_window_id: ctx.agentWindowId, + drop_only: options.dropOnly ?? false, + live_session_ids: this.list().map((entry) => entry.sessionId), + }); return ctx; } diff --git a/apps/extension/src/tools/dispatcher.ts b/apps/extension/src/tools/dispatcher.ts index e0a5ac4..948d554 100644 --- a/apps/extension/src/tools/dispatcher.ts +++ b/apps/extension/src/tools/dispatcher.ts @@ -1,3 +1,4 @@ +import type { DiagnosticSink } from "@/lib/diagnostics"; import { OVERLAY_AUTOMATION_BYPASS } from "@/lib/overlay-bridge"; import type { SessionManager } from "@/session-manager/manager"; import type { Transport } from "@/transport/transport"; @@ -118,6 +119,7 @@ export interface DispatcherDeps { approveBorrow?: BorrowConfirmationApprover; /** i18n notification copy for `tool.request_help` (resolved per-call). */ helpNotificationCopy?: () => { title: string; body: string }; + diagnostics?: DiagnosticSink; } /** @@ -145,6 +147,7 @@ export class ToolDispatcher { private readonly onBrowserControlResumed?: (sessionId: string) => void; private readonly approveBorrow?: BorrowConfirmationApprover; private readonly helpNotificationCopy?: () => { title: string; body: string }; + private readonly diagnostics?: DiagnosticSink; private subscription: { dispose(): void } | null = null; private readonly hoverBypassTabs = new Map(); private readonly hoverLatches = new Map(); @@ -165,6 +168,7 @@ export class ToolDispatcher { this.onBrowserControlResumed = deps.onBrowserControlResumed; this.approveBorrow = deps.approveBorrow; this.helpNotificationCopy = deps.helpNotificationCopy; + this.diagnostics = deps.diagnostics; } start(): void { @@ -192,6 +196,16 @@ export class ToolDispatcher { private async dispatch(msg: ProtocolFrame): Promise { if (!isRequestFrame(msg)) return; const req = msg as RequestFrame; + const requestSessionId = (req.params as { session_id?: unknown } | undefined)?.session_id; + const startedAt = performance.now(); + this.diagnostics?.("dispatcher.rpc.received", { + rpc_id: req.id, + method: req.method, + session_id: typeof requestSessionId === "string" ? requestSessionId : null, + transport_state: this.transport.state, + live_session_ids: this.sessions.list().map((ctx) => ctx.sessionId), + inflight_rpc_ids: Array.from(this.inflightAbortControllers.keys()), + }); // Cancel frames take a fast path: trip the matching controller // (if any), reply with `{cancelled}` so the daemon can answer @@ -200,6 +214,12 @@ export class ToolDispatcher { const params = (req.params as { rpc_id?: string } | undefined) ?? {}; const target = typeof params.rpc_id === "string" ? params.rpc_id : ""; const ac = target ? this.inflightAbortControllers.get(target) : undefined; + this.diagnostics?.("dispatcher.cancel.received", { + rpc_id: req.id, + target_rpc_id: target, + target_found: ac !== undefined, + inflight_rpc_ids: Array.from(this.inflightAbortControllers.keys()), + }); if (ac) { try { ac.abort(); @@ -213,7 +233,17 @@ export class ToolDispatcher { }; try { this.transport.send(reply); + this.diagnostics?.("dispatcher.cancel.ack_sent", { + rpc_id: req.id, + target_rpc_id: target, + cancelled: ac !== undefined, + }); } catch (sendErr) { + this.diagnostics?.("dispatcher.cancel.ack_failed", { + rpc_id: req.id, + target_rpc_id: target, + error: sendErr, + }); console.warn("[bsk dispatcher] failed to ack cancel", sendErr); } return; @@ -238,6 +268,13 @@ export class ToolDispatcher { } } } catch (err) { + this.diagnostics?.("dispatcher.rpc.invoke_failed", { + rpc_id: req.id, + method: req.method, + session_id: typeof requestSessionId === "string" ? requestSessionId : null, + abort_signal: ac.signal.aborted, + error: err, + }); if (isAbortLikeError(err)) { body = { id: req.id, @@ -258,8 +295,27 @@ export class ToolDispatcher { let sent = true; try { this.transport.send(body); + this.diagnostics?.("dispatcher.rpc.response_sent", { + rpc_id: req.id, + method: req.method, + session_id: typeof requestSessionId === "string" ? requestSessionId : null, + response_kind: "error" in body ? "error" : "ok", + error_code: "error" in body ? body.error.code : null, + elapsed_ms: Math.round(performance.now() - startedAt), + transport_state: this.transport.state, + live_session_ids: this.sessions.list().map((ctx) => ctx.sessionId), + }); } catch (sendErr) { sent = false; + this.diagnostics?.("dispatcher.rpc.response_send_failed", { + rpc_id: req.id, + method: req.method, + session_id: typeof requestSessionId === "string" ? requestSessionId : null, + started_session_id: startedSession, + elapsed_ms: Math.round(performance.now() - startedAt), + transport_state: this.transport.state, + error: sendErr, + }); // Transport is dead by the time we want to reply. Drop the link // proactively so the alarm-driven keepalive reconnects sooner // and the daemon's pending RPC times out cleanly instead of @@ -278,12 +334,23 @@ export class ToolDispatcher { try { const ctx = await this.sessions.stop(startedSession); if (ctx) { + this.diagnostics?.("dispatcher.session_start.rolled_back", { + rpc_id: req.id, + session_id: startedSession, + agent_window_id: ctx.agentWindowId, + reason: "response_send_failed", + }); console.warn( "[bsk dispatcher] rolled back orphan session after send failure", startedSession, ); } } catch (rollbackErr) { + this.diagnostics?.("dispatcher.session_start.rollback_failed", { + rpc_id: req.id, + session_id: startedSession, + error: rollbackErr, + }); console.warn("[bsk dispatcher] session rollback after send failure failed", rollbackErr); } } diff --git a/apps/extension/src/transport/ws-transport.ts b/apps/extension/src/transport/ws-transport.ts index 2f57782..50bd08e 100644 --- a/apps/extension/src/transport/ws-transport.ts +++ b/apps/extension/src/transport/ws-transport.ts @@ -1,3 +1,4 @@ +import type { DiagnosticSink } from "@/lib/diagnostics"; import type { ConnectionStateHandler, Disposable, FrameHandler, Transport } from "./transport"; import type { ConnectionState, ProtocolFrame } from "./types"; @@ -18,6 +19,7 @@ export interface WSTransportOptions { * Inject a fake WebSocket constructor in tests; defaults to `globalThis.WebSocket`. */ webSocketFactory?: WebSocketFactory; + diagnostics?: DiagnosticSink; } interface CloseLikeEvent { @@ -44,6 +46,7 @@ export class WSTransport implements Transport { private readonly factory: WebSocketFactory; private readonly initialDelayMs: number; private readonly maxDelayMs: number; + private readonly diagnostics?: DiagnosticSink; private socket: WebSocket | null = null; private currentState: ConnectionState = "disconnected"; @@ -63,6 +66,7 @@ export class WSTransport implements Transport { this.factory = options.webSocketFactory ?? ((url: string) => new WebSocket(url)); this.initialDelayMs = options.reconnect?.initialDelayMs ?? DEFAULT_INITIAL_DELAY_MS; this.maxDelayMs = options.reconnect?.maxDelayMs ?? DEFAULT_MAX_DELAY_MS; + this.diagnostics = options.diagnostics; } get state(): ConnectionState { @@ -70,6 +74,13 @@ export class WSTransport implements Transport { } connect(): Promise { + this.diagnostics?.("transport.connect.requested", { + state: this.currentState, + socket_generation: this.socketGeneration, + has_socket: this.socket !== null, + has_connecting_promise: this.connectingPromise !== null, + has_reconnect_timer: this.reconnectTimer !== null, + }); if (this.currentState === "connected") return Promise.resolve(); this.explicitlyClosed = false; @@ -94,6 +105,11 @@ export class WSTransport implements Transport { } async disconnect(): Promise { + this.diagnostics?.("transport.disconnect.requested", { + state: this.currentState, + socket_generation: this.socketGeneration, + has_socket: this.socket !== null, + }); this.explicitlyClosed = true; if (this.reconnectTimer) { clearTimeout(this.reconnectTimer); @@ -124,6 +140,12 @@ export class WSTransport implements Transport { send(msg: ProtocolFrame): void { if (!this.socket || this.socket.readyState !== 1 /* OPEN */) { + this.diagnostics?.("transport.send.rejected", { + state: this.currentState, + socket_generation: this.socketGeneration, + socket_ready_state: this.socket?.readyState ?? null, + frame_kind: "method" in msg ? msg.method : "event" in msg ? msg.event : "response", + }); throw new Error("[WSTransport] cannot send while not connected"); } this.socket.send(JSON.stringify(msg)); @@ -152,10 +174,16 @@ export class WSTransport implements Transport { const generation = ++this.socketGeneration; const socket = this.factory(this.url); this.socket = socket; + this.diagnostics?.("transport.socket.created", { + socket_generation: generation, + reconnect_attempt: this.reconnectAttempt, + url: this.url, + }); socket.addEventListener("open", () => { if (!this.isCurrentSocket(socket, generation)) return; this.reconnectAttempt = 0; + this.diagnostics?.("transport.socket.open", { socket_generation: generation }); this.setState("connected"); const resolve = this.resolveConnect; this.resolveConnect = null; @@ -174,6 +202,7 @@ export class WSTransport implements Transport { }); socket.addEventListener("error", () => { + this.diagnostics?.("transport.socket.error", { socket_generation: generation }); // The 'close' handler always fires after 'error' in browsers, so we // just record the error and let close drive reconnect logic. }); @@ -197,7 +226,16 @@ export class WSTransport implements Transport { } private handleClose(socket: WebSocket, generation: number, _ev: CloseLikeEvent): void { - if (!this.isCurrentSocket(socket, generation)) return; + const isCurrent = this.isCurrentSocket(socket, generation); + this.diagnostics?.("transport.socket.close", { + socket_generation: generation, + current_socket_generation: this.socketGeneration, + is_current: isCurrent, + explicitly_closed: this.explicitlyClosed, + close_code: _ev.code ?? null, + close_reason: _ev.reason ?? "", + }); + if (!isCurrent) return; this.socket = null; if (this.explicitlyClosed) { this.setState("disconnected"); @@ -211,9 +249,18 @@ export class WSTransport implements Transport { if (this.reconnectTimer) return; const delay = Math.min(this.initialDelayMs * 2 ** this.reconnectAttempt, this.maxDelayMs); this.reconnectAttempt += 1; + this.diagnostics?.("transport.reconnect.scheduled", { + delay_ms: delay, + reconnect_attempt: this.reconnectAttempt, + socket_generation: this.socketGeneration, + }); this.reconnectTimer = setTimeout(() => { this.reconnectTimer = null; if (this.explicitlyClosed) return; + this.diagnostics?.("transport.reconnect.timer_fired", { + reconnect_attempt: this.reconnectAttempt, + socket_generation: this.socketGeneration, + }); void this.connect().catch((err) => { console.debug("[WSTransport] reconnect attempt failed", err); }); @@ -226,7 +273,13 @@ export class WSTransport implements Transport { private setState(next: ConnectionState): void { if (this.currentState === next) return; + const previous = this.currentState; this.currentState = next; + this.diagnostics?.("transport.state.changed", { + previous, + next, + socket_generation: this.socketGeneration, + }); for (const h of this.stateHandlers) { try { h(next); diff --git a/apps/extension/wxt.config.ts b/apps/extension/wxt.config.ts index 6e272d9..47f3e79 100644 --- a/apps/extension/wxt.config.ts +++ b/apps/extension/wxt.config.ts @@ -22,7 +22,8 @@ export default defineConfig({ outDir: "dist", modules: ["@wxt-dev/module-react"], manifest: { - name: "BrowserSkill", + name: "BrowserSkill (Lumi Diagnostics)", + version_name: `${EXTENSION_VERSION}-lumi-session-loss-v1`, description: "Let AI agents use your logged-in browser in a separate Agent Window—without interrupting your work. Powered by the bsk CLI.", // Flat debugger sessions are required to address out-of-process iframes. @@ -45,7 +46,7 @@ export default defineConfig({ 128: "icon/logo.png", }, action: { - default_title: "BrowserSkill", + default_title: "BrowserSkill (Lumi Diagnostics)", default_icon: { 16: "icon/logo.png", 32: "icon/logo.png", diff --git a/crates/bsk-cli/src/daemon/browsers.rs b/crates/bsk-cli/src/daemon/browsers.rs index b1dad44..a67fbd2 100644 --- a/crates/bsk-cli/src/daemon/browsers.rs +++ b/crates/bsk-cli/src/daemon/browsers.rs @@ -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> { + 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> { diff --git a/crates/bsk-cli/src/daemon/ipc.rs b/crates/bsk-cli/src/daemon/ipc.rs index 38d0046..2ab7d28 100644 --- a/crates/bsk-cli/src/daemon/ipc.rs +++ b/crates/bsk-cli/src/daemon/ipc.rs @@ -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) -> 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) -> 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, params: Value) -> ResponseBody { }; let local = state.abort_registry.cancel(¶ms.rpc_id); let mut cancelled = local; + let mut surface = if local { "abort_registry" } else { "none" }; if !local && let Some(snap) = state.tool_inflight.cancel(¶ms.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, 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)) } diff --git a/crates/bsk-cli/src/daemon/queue.rs b/crates/bsk-cli/src/daemon/queue.rs index 05f09e8..bfdefb1 100644 --- a/crates/bsk-cli/src/daemon/queue.rs +++ b/crates/bsk-cli/src/daemon/queue.rs @@ -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, job: &ToolJob, ) -> Result { + 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), diff --git a/crates/bsk-cli/src/daemon/sessions.rs b/crates/bsk-cli/src/daemon/sessions.rs index 7eab0e5..2baa0a7 100644 --- a/crates/bsk-cli/src/daemon/sessions.rs +++ b/crates/bsk-cli/src/daemon/sessions.rs @@ -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 { + 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 { + let caller = std::panic::Location::caller(); let mut guard = self.inner.lock().expect("session registry poisoned"); let drained: Vec = 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 } diff --git a/crates/bsk-cli/src/daemon/start.rs b/crates/bsk-cli/src/daemon/start.rs index 241eb85..35a8ca0 100644 --- a/crates/bsk-cli/src/daemon/start.rs +++ b/crates/bsk-cli/src/daemon/start.rs @@ -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 { 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"); diff --git a/crates/bsk-cli/src/daemon/ws.rs b/crates/bsk-cli/src/daemon/ws.rs index 435b0a9..5d3f5ba 100644 --- a/crates/bsk-cli/src/daemon/ws.rs +++ b/crates/bsk-cli/src/daemon/ws.rs @@ -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 = 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, client: &Arc, text: &str) { @@ -442,6 +469,38 @@ async fn handle_inbound_text(state: &Arc, client: &Arc { + 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); } diff --git a/crates/bsk-protocol/src/frame.rs b/crates/bsk-protocol/src/frame.rs index 996d487..d2ab4d1 100644 --- a/crates/bsk-protocol/src/frame.rs +++ b/crates/bsk-protocol/src/frame.rs @@ -58,6 +58,10 @@ pub enum EventKind { BrowserDisconnected, #[serde(rename = "browser.connected")] BrowserConnected, + /// Structured diagnostics emitted by locally instrumented extension + /// builds. Production peers may ignore this event safely. + #[serde(rename = "browser.diagnostic")] + BrowserDiagnostic, } #[derive(Debug, Clone, PartialEq)] @@ -256,6 +260,27 @@ mod tests { assert_eq!(frame.event, EventKind::SystemHeartbeat); } + #[test] + fn browser_diagnostic_event_frame_round_trips_from_extension_shape() { + let wire = serde_json::json!({ + "event": "browser.diagnostic", + "payload": { + "schema_version": 1, + "worker_boot_id": "worker-1", + "entries": [{ "event": "extension.worker.started" }] + } + }); + let frame: EventFrame = serde_json::from_value(wire).unwrap(); + assert_eq!(frame.event, EventKind::BrowserDiagnostic); + assert_eq!( + frame + .payload + .get("worker_boot_id") + .and_then(|value| value.as_str()), + Some("worker-1") + ); + } + #[test] fn session_user_interrupt_serialises_as_snake_case() { let v = serde_json::to_value(EventKind::SessionUserInterrupt).unwrap();