From a7bc8bacf6e912dc009a87b32799b496162fe244 Mon Sep 17 00:00:00 2001 From: BB-fat <1056871944@qq.com> Date: Fri, 14 Aug 2026 04:59:26 +0000 Subject: [PATCH] feat(dsh-plugin-browserskill): host-side ObservationService for the PiP overlay (Phase 1-A) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Per-session live observation for owned sessions only, feeding the upcoming client overlay: - State model: SessionObservation (sessionId/url/action/since/ thumbnailAttachmentId/lastError), one entry per owned session; added on start, removed on stop, cleared on dispose. - Instrumentation: the shared runBsk wrapper begins/ends an action (tool name mapped to a verb) around every model-facing call and tags the child with its session id; action end triggers an immediate thumbnail refresh. - Thumbnail loop: 1.5s cadence while active, 8s idle downclock, x3 failure backoff, silent frame retention on errors; captures run OUTSIDE the instrumentation (no action events, no current-pointer movement) and land in the attachment store by reference. - Interrupt: runner.killFor(tag) kills exactly the tagged in-flight bsk children — chat-Stop-equivalent semantics for the current or a named owned session; foreign ids and the no-in-flight case return false. - Remote seam: dsh 0.1's Typert Remote pipeline and forwarded-event allowlist are closed to out-of-tree packages, so state/events/ interrupt are served over the documented webServer route seam: GET /bsk-observation/state, GET /bsk-observation/events (SSE), POST /bsk-observation/interrupt; routes mount only when a webServer service exists (headless compositions skip them). - Config: observationEnabled (true), thumbnailIntervalMs (1500), idleIntervalMs (8000). Tests (+13, 72 total): state machine + event stream, cadence fast/idle/ backoff, headless no-store silence, observation-traffic isolation, interrupt routing (default/specified/none/foreign), HTTP route handlers, disabled-webServer no-op. --- packages/dsh-plugin-browserskill/src/index.ts | 32 +- .../src/observation-http.ts | 116 +++++ .../src/observation.ts | 320 ++++++++++++++ .../dsh-plugin-browserskill/src/runner.ts | 25 +- packages/dsh-plugin-browserskill/src/tools.ts | 78 +++- .../tests/dispose.test.ts | 1 + .../tests/observation.test.ts | 407 ++++++++++++++++++ .../tests/tools.test.ts | 41 +- 8 files changed, 989 insertions(+), 31 deletions(-) create mode 100644 packages/dsh-plugin-browserskill/src/observation-http.ts create mode 100644 packages/dsh-plugin-browserskill/src/observation.ts create mode 100644 packages/dsh-plugin-browserskill/tests/observation.test.ts diff --git a/packages/dsh-plugin-browserskill/src/index.ts b/packages/dsh-plugin-browserskill/src/index.ts index ab48da1..28c31f5 100644 --- a/packages/dsh-plugin-browserskill/src/index.ts +++ b/packages/dsh-plugin-browserskill/src/index.ts @@ -12,6 +12,8 @@ import type { Context } from "@deepseek-ai/cordis"; import Schema from "@deepseek-ai/schemastery"; +import { ObservationService } from "./observation"; +import { registerObservationRoutes } from "./observation-http"; import { type BskRunner, createBskRunner } from "./runner"; import { SessionRegistry } from "./sessions"; import { type PluginConfig, registerTools } from "./tools"; @@ -30,6 +32,15 @@ export const Config = Schema.object({ maxSessions: Schema.number() .default(5) .description("Maximum number of concurrent browser sessions started through this plugin."), + observationEnabled: Schema.boolean() + .default(true) + .description("Track per-session observation state (action/url/thumbnail) for the PiP overlay."), + thumbnailIntervalMs: Schema.number() + .default(1500) + .description("Thumbnail refresh cadence for active sessions (milliseconds)."), + idleIntervalMs: Schema.number() + .default(8000) + .description("Thumbnail refresh cadence for idle sessions; also the recent-activity window."), }); export type Config = PluginConfig; @@ -48,11 +59,25 @@ export function apply( bskPath: config.bskPath ?? "bsk", defaultTimeoutMs: config.defaultTimeoutMs ?? 120_000, maxSessions: config.maxSessions ?? 5, + observationEnabled: config.observationEnabled ?? true, + thumbnailIntervalMs: config.thumbnailIntervalMs ?? 1500, + idleIntervalMs: config.idleIntervalMs ?? 8000, }; const runner = options.runnerFactory?.(resolved.bskPath) ?? createBskRunner(resolved.bskPath); const registry = new SessionRegistry(resolved.maxSessions); + const observation = new ObservationService({ + ctx, + runner, + registry, + options: { + enabled: resolved.observationEnabled, + thumbnailIntervalMs: resolved.thumbnailIntervalMs, + idleIntervalMs: resolved.idleIntervalMs, + }, + }); - registerTools({ ctx, runner, registry, config: resolved }); + registerTools({ ctx, runner, registry, config: resolved, observation }); + const removeRoutes = registerObservationRoutes(ctx, observation); // Non-blocking install probe: warn early when bsk is missing instead of // failing the first tool call with a bare spawn error. @@ -73,6 +98,8 @@ export function apply( // are swallowed so one stale handle cannot abort the rest. ctx.effect(() => { return () => { + removeRoutes(); + observation.dispose(); runner.killAll(); const stops = registry .ownedIds() @@ -84,6 +111,9 @@ export function apply( }); } +export type { ObservationEvent, ObservationOptions, SessionObservation } from "./observation"; +export { ObservationService } from "./observation"; +export { registerObservationRoutes } from "./observation-http"; export type { BskRunner, BskRunResult, SpawnImpl } from "./runner"; export { BskError, createBskRunner } from "./runner"; export { SessionRegistry } from "./sessions"; diff --git a/packages/dsh-plugin-browserskill/src/observation-http.ts b/packages/dsh-plugin-browserskill/src/observation-http.ts new file mode 100644 index 0000000..343c96e --- /dev/null +++ b/packages/dsh-plugin-browserskill/src/observation-http.ts @@ -0,0 +1,116 @@ +/** + * HTTP/SSE exposure of the ObservationService for the client half. The Typert + * Remote pipeline and the forwarded-event allowlist are both closed to + * out-of-tree packages in dsh 0.1 (generation runs over dsh's own ts.Program; + * the event allowlist lives in dsh-api-remotes), so the plugin serves its + * observation channel through the documented `webServer` route seam instead: + * + * GET /bsk-observation/state → { sessions: SessionObservation[] } + * GET /bsk-observation/events → SSE stream of ObservationEvent + * POST /bsk-observation/interrupt → body {sessionId?} → {interrupted: boolean} + * + * Routes exist only when a webServer service is mounted (web composition); + * headless deployments skip registration entirely. + */ + +import type { IncomingMessage, ServerResponse } from "node:http"; +import type { Context } from "@deepseek-ai/cordis"; +import type { ObservationService } from "./observation"; + +/** Structural view of the dsh-host-webserver route seam. */ +interface WebServerLike { + register(route: { + kind: "exact" | "prefix"; + path: string; + handler: (req: IncomingMessage, res: ServerResponse) => void | Promise; + }): () => void; +} + +const ROUTE_BASE = "/bsk-observation"; +const SSE_HEARTBEAT_MS = 15_000; + +function sendJson(res: ServerResponse, status: number, body: unknown): void { + res.writeHead(status, { "content-type": "application/json; charset=utf-8" }); + res.end(JSON.stringify(body)); +} + +/** + * Register the observation routes. No-op (with a console note) when the + * composition has no web server. + * @returns disposer removing the routes. + */ +export function registerObservationRoutes( + ctx: Context, + observation: ObservationService, +): () => void { + const webServer = ctx.get("webServer") as WebServerLike | undefined; + if (webServer === undefined) { + return () => {}; + } + const disposers = [ + webServer.register({ + kind: "exact", + path: `${ROUTE_BASE}/state`, + handler: (req, res) => { + if (req.method !== "GET") { + sendJson(res, 405, { error: "method not allowed" }); + return; + } + sendJson(res, 200, { sessions: observation.getState() }); + }, + }), + webServer.register({ + kind: "exact", + path: `${ROUTE_BASE}/events`, + handler: (req, res) => { + if (req.method !== "GET") { + sendJson(res, 405, { error: "method not allowed" }); + return; + } + res.writeHead(200, { + "content-type": "text/event-stream", + "cache-control": "no-cache", + connection: "keep-alive", + }); + const unsubscribe = observation.subscribe((event) => { + res.write(`data: ${JSON.stringify(event)}\n\n`); + }); + const heartbeat = setInterval(() => res.write(": heartbeat\n\n"), SSE_HEARTBEAT_MS); + req.on("close", () => { + clearInterval(heartbeat); + unsubscribe(); + }); + }, + }), + webServer.register({ + kind: "exact", + path: `${ROUTE_BASE}/interrupt`, + handler: (req, res) => { + if (req.method !== "POST") { + sendJson(res, 405, { error: "method not allowed" }); + return; + } + let body = ""; + req.on("data", (chunk: Buffer | string) => { + body += chunk; + }); + req.on("end", () => { + let sessionId: string | undefined; + try { + const parsed = JSON.parse(body || "{}") as { sessionId?: unknown }; + if (typeof parsed.sessionId === "string" && parsed.sessionId !== "") { + sessionId = parsed.sessionId; + } + } catch { + sendJson(res, 400, { error: "invalid JSON body" }); + return; + } + sendJson(res, 200, { interrupted: observation.interrupt(sessionId) }); + }); + }, + }), + ]; + return () => { + for (const dispose of disposers) dispose(); + }; +} diff --git a/packages/dsh-plugin-browserskill/src/observation.ts b/packages/dsh-plugin-browserskill/src/observation.ts new file mode 100644 index 0000000..1bcf338 --- /dev/null +++ b/packages/dsh-plugin-browserskill/src/observation.ts @@ -0,0 +1,320 @@ +/** + * ObservationService: per-owned-session live observation state for the PiP + * overlay — current action, page url, and a breathing thumbnail carried by + * attachment reference. Strict ownership boundary applies throughout: only + * sessions in the plugin's SessionRegistry (owned by construction) ever get + * an observation entry, and interrupt/kill paths can only reach children this + * plugin spawned. + * + * Observation traffic isolation: thumbnail captures run through the runner + * directly (never through the tool-level instrumentation), emit no action + * events, and never move the registry's current pointer. + */ + +import { readFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import type { Context } from "@deepseek-ai/cordis"; +import type { BskRunner } from "./runner"; +import type { SessionRegistry } from "./sessions"; + +/** One owned session's live observation record (wire-stable shape). */ +export interface SessionObservation { + sessionId: string; + /** Last settled page URL, when any navigation completed. */ + url?: string; + /** Current action ('idle' when nothing is in flight). */ + action: string; + /** Epoch ms when the current action started (elapsed time is client-side). */ + since: number; + /** Latest thumbnail, by attachment-store reference (never bytes on the wire). */ + thumbnailAttachmentId?: string; + /** Summary of the most recent failed action (drives the red status dot). */ + lastError?: string; +} + +/** Incremental event carried to subscribers (SSE on the wire). */ +export interface ObservationEvent { + type: "upsert" | "remove" | "reset"; + session?: SessionObservation; +} + +export interface ObservationOptions { + enabled: boolean; + /** Fast cadence while a session is active or recently was. */ + thumbnailIntervalMs: number; + /** Slow cadence for idle sessions; also the "recently active" window. */ + idleIntervalMs: number; +} + +/** Structural view of the optional attachment seam (absent in headless compositions). */ +interface AttachmentLike { + saveImage(input: { + data: Uint8Array; + mediaType: string; + name?: string; + }): Promise<{ attachmentId: unknown }>; +} + +/** Injectable clock/scheduler bits for tests. */ +export interface ObservationScheduler { + setTimeout: (fn: () => void, ms: number) => unknown; + clearTimeout: (handle: unknown) => void; + now: () => number; +} + +const DEFAULT_SCHEDULER: ObservationScheduler = { + setTimeout: (fn, ms) => { + const timer = setTimeout(fn, ms); + // Never hold the event loop open for a frame refresh. + if (typeof timer === "object" && typeof timer.unref === "function") timer.unref(); + return timer; + }, + clearTimeout: (h) => clearTimeout(h as Parameters[0]), + now: () => Date.now(), +}; + +/** Max consecutive capture failures before a session drops to the idle cadence. */ +const FAILURE_BACKOFF_THRESHOLD = 3; + +export class ObservationService { + private readonly observations = new Map(); + private readonly listeners = new Set<(event: ObservationEvent) => void>(); + private readonly captureTimers = new Map(); + private readonly captureInFlight = new Set(); + private readonly captureFailures = new Map(); + private readonly lastActivity = new Map(); + private disposed = false; + + constructor( + private readonly deps: { + ctx: Context; + runner: BskRunner; + registry: SessionRegistry; + options: ObservationOptions; + scheduler?: ObservationScheduler; + }, + ) {} + + private get scheduler(): ObservationScheduler { + return this.deps.scheduler ?? DEFAULT_SCHEDULER; + } + + /** All current entries (client initial/resync snapshot). */ + getState(): SessionObservation[] { + return [...this.observations.values()].map((entry) => ({ ...entry })); + } + + /** Subscribe to incremental changes; returns an unsubscribe function. */ + subscribe(listener: (event: ObservationEvent) => void): () => void { + this.listeners.add(listener); + return () => this.listeners.delete(listener); + } + + private emit(event: ObservationEvent): void { + for (const listener of [...this.listeners]) { + try { + listener(event); + } catch { + // A faulty subscriber must not starve the others. + } + } + } + + private put(entry: SessionObservation): void { + this.observations.set(entry.sessionId, entry); + this.emit({ type: "upsert", session: { ...entry } }); + } + + /** Register a fresh owned session (called from browser_session_start). */ + addSession(sessionId: string, url?: string): void { + if (!this.deps.options.enabled || this.disposed) return; + this.put({ + sessionId, + ...(url !== undefined ? { url } : {}), + action: "idle", + since: this.scheduler.now(), + }); + this.lastActivity.set(sessionId, this.scheduler.now()); + this.scheduleCapture(sessionId, 0); + } + + /** Drop a session (called from browser_session_stop). */ + removeSession(sessionId: string): void { + this.cancelCapture(sessionId); + this.captureFailures.delete(sessionId); + this.lastActivity.delete(sessionId); + if (this.observations.delete(sessionId)) { + this.emit({ + type: "remove", + session: { sessionId, action: "idle", since: this.scheduler.now() }, + }); + } + } + + /** Mark an action starting on a session (tool entry instrumentation). */ + beginAction(sessionId: string, action: string): void { + if (!this.deps.options.enabled || this.disposed) return; + const entry = this.observations.get(sessionId); + if (entry === undefined) return; + const now = this.scheduler.now(); + this.lastActivity.set(sessionId, now); + this.put({ ...entry, action, since: now }); + } + + /** + * Mark the current action settling (tool exit instrumentation). Triggers an + * immediate thumbnail refresh — action-driven first, timer as fallback. + */ + endAction(sessionId: string, error?: string): void { + if (!this.deps.options.enabled || this.disposed) return; + const entry = this.observations.get(sessionId); + if (entry === undefined) return; + const now = this.scheduler.now(); + this.lastActivity.set(sessionId, now); + this.put({ + ...entry, + action: "idle", + since: now, + ...(error !== undefined ? { lastError: error } : {}), + }); + this.scheduleCapture(sessionId, 0); + } + + /** Record the settled page URL (navigate success / start with url). */ + setUrl(sessionId: string, url: string): void { + if (!this.deps.options.enabled || this.disposed) return; + const entry = this.observations.get(sessionId); + if (entry === undefined) return; + this.put({ ...entry, url }); + } + + /** + * Interrupt the in-flight call of one session (default: the registry's + * current). Kills exactly the bsk children this plugin spawned for that + * session — same user-visible semantics as the chat Stop button (the in-flight + * tool call fails; the agent flow may continue). + * @returns whether an in-flight call was actually interrupted. + */ + interrupt(sessionId?: string): boolean { + const target = sessionId ?? this.deps.registry.current(); + if (target === undefined || !this.deps.registry.isOwned(target)) return false; + return this.deps.runner.killFor(target) > 0; + } + + /** Tear down all state and timers (plugin dispose). */ + dispose(): void { + this.disposed = true; + for (const sessionId of [...this.captureTimers.keys()]) this.cancelCapture(sessionId); + this.captureTimers.clear(); + this.captureInFlight.clear(); + this.captureFailures.clear(); + this.lastActivity.clear(); + this.observations.clear(); + this.emit({ type: "reset" }); + this.listeners.clear(); + } + + // ------------------------------------------------------------------ + // Thumbnail loop + // ------------------------------------------------------------------ + + private cancelCapture(sessionId: string): void { + const timer = this.captureTimers.get(sessionId); + if (timer !== undefined) { + this.scheduler.clearTimeout(timer); + this.captureTimers.delete(sessionId); + } + } + + /** Schedule the next capture for a session; `delayMs` 0 means "as soon as the event loop allows". */ + private scheduleCapture(sessionId: string, delayMs?: number): void { + if (!this.deps.options.enabled || this.disposed) return; + if (!this.observations.has(sessionId)) return; + this.cancelCapture(sessionId); + const now = this.scheduler.now(); + const lastSeen = this.lastActivity.get(sessionId) ?? 0; + const failures = this.captureFailures.get(sessionId) ?? 0; + const { thumbnailIntervalMs, idleIntervalMs } = this.deps.options; + const cadence = + now - lastSeen < idleIntervalMs && failures < FAILURE_BACKOFF_THRESHOLD + ? thumbnailIntervalMs + : idleIntervalMs; + const delay = delayMs ?? cadence; + const timer = this.scheduler.setTimeout(() => { + this.captureTimers.delete(sessionId); + void this.capture(sessionId); + }, delay); + this.captureTimers.set(sessionId, timer); + } + + /** + * Capture one frame: `bsk screenshot --json` through the runner, bytes into + * the attachment store, reference onto the observation. Runs OUTSIDE the + * tool instrumentation on purpose — no action events, no registry writes. + * Failures keep the previous frame and back off silently. + */ + private async capture(sessionId: string): Promise { + if (this.disposed || !this.observations.has(sessionId)) return; + if (this.captureInFlight.has(sessionId)) return; + const attachments = this.deps.ctx.get("attachments") as AttachmentLike | undefined; + if (attachments === undefined) { + // No attachment store (headless composition): thumbnails stay absent. + return; + } + this.captureInFlight.add(sessionId); + try { + const outPath = join(tmpdir(), `bsk-obs-${sessionId}-${this.scheduler.now()}.png`); + const result = await this.deps.runner.run( + ["screenshot", "--session", sessionId, "--out", outPath], + { timeoutMs: 15_000, tag: `observation:${sessionId}` }, + ); + if (result.code !== 0) throw new Error(`screenshot exited ${result.code}`); + const reply = JSON.parse(result.stdout) as { path?: string }; + const data = await readFile(reply.path ?? outPath); + const ref = await attachments.saveImage({ + data, + mediaType: "image/png", + name: `observation-${sessionId}.png`, + }); + const entry = this.observations.get(sessionId); + if (entry === undefined) return; + this.captureFailures.delete(sessionId); + this.put({ ...entry, thumbnailAttachmentId: String(ref.attachmentId) }); + } catch { + // Silent by design: keep the previous frame, count toward backoff. + this.captureFailures.set(sessionId, (this.captureFailures.get(sessionId) ?? 0) + 1); + } finally { + this.captureInFlight.delete(sessionId); + if (!this.disposed && this.observations.has(sessionId)) this.scheduleCapture(sessionId); + } + } +} + +/** Map a bsk command label onto its observation action verb. */ +export function actionForLabel(label: string): string { + switch (label) { + case "session start": + return "starting"; + case "session stop": + return "stopping"; + case "navigate": + return "navigating"; + case "snapshot": + return "snapshotting"; + case "observe": + return "observing"; + case "click": + return "clicking"; + case "fill": + return "filling"; + case "press": + return "pressing"; + case "screenshot": + return "capturing"; + case "emulate": + return "emulating"; + default: + return label; + } +} diff --git a/packages/dsh-plugin-browserskill/src/runner.ts b/packages/dsh-plugin-browserskill/src/runner.ts index c747177..7a2dca7 100644 --- a/packages/dsh-plugin-browserskill/src/runner.ts +++ b/packages/dsh-plugin-browserskill/src/runner.ts @@ -53,17 +53,26 @@ export interface BskRunOptions { /** Minimal spawn signature so tests can substitute a fake child process. */ export type SpawnImpl = (command: string, args: string[]) => ChildProcess; +export interface BskRunOptions { + signal?: AbortSignal; + timeoutMs?: number; + /** Opaque routing tag (e.g. a session id) enabling per-tag kills. */ + tag?: string; +} + export interface BskRunner { /** Run `bsk --json` and collect its output. */ run(args: string[], options?: BskRunOptions): Promise; /** Kill every in-flight child (used when the plugin unloads). */ killAll(): void; + /** Kill only the in-flight children carrying this tag; returns how many were killed. */ + killFor(tag: string): number; } const KILL_GRACE_MS = 2000; export function createBskRunner(bskPath: string, spawnImpl: SpawnImpl = spawn): BskRunner { - const live = new Set(); + const live = new Map(); function killChild(child: ChildProcess): void { if (child.exitCode !== null || child.signalCode !== null) return; @@ -84,7 +93,7 @@ export function createBskRunner(bskPath: string, spawnImpl: SpawnImpl = spawn): reject(error); return; } - live.add(child); + live.set(child, options.tag); let stdout = ""; let stderr = ""; @@ -134,7 +143,17 @@ export function createBskRunner(bskPath: string, spawnImpl: SpawnImpl = spawn): }); }, killAll() { - for (const child of live) killChild(child); + for (const child of live.keys()) killChild(child); + }, + killFor(tag: string) { + let killed = 0; + for (const [child, childTag] of live) { + if (childTag === tag) { + killChild(child); + killed += 1; + } + } + return killed; }, }; } diff --git a/packages/dsh-plugin-browserskill/src/tools.ts b/packages/dsh-plugin-browserskill/src/tools.ts index 869789f..d78e8c3 100644 --- a/packages/dsh-plugin-browserskill/src/tools.ts +++ b/packages/dsh-plugin-browserskill/src/tools.ts @@ -13,6 +13,7 @@ import type { Context } from "@deepseek-ai/cordis"; import type { ContentBlock } from "@deepseek-ai/dsh-llm"; import { defineTool, type ToolResult, type ToolRunContext } from "@deepseek-ai/dsh-tools"; import { trySaveScreenshot } from "./image"; +import { actionForLabel, type ObservationService } from "./observation"; import { type BskRunner, bskInstallMessage, isCommandNotFound, parseBskJson } from "./runner"; import type { SessionRegistry } from "./sessions"; @@ -21,6 +22,9 @@ export interface PluginConfig { bskPath: string; defaultTimeoutMs: number; maxSessions: number; + observationEnabled: boolean; + thumbnailIntervalMs: number; + idleIntervalMs: number; } export interface ToolDeps { @@ -28,6 +32,7 @@ export interface ToolDeps { runner: BskRunner; registry: SessionRegistry; config: PluginConfig; + observation: ObservationService; } /** Device presets supported by `bsk emulate --device`. */ @@ -48,31 +53,52 @@ const SESSION_PARAM = { "Omit to use the current session (the one most recently started or used).", } as const; -/** Run one bsk command and return its parsed JSON payload, or throw. */ +/** + * Run one bsk command and return its parsed JSON payload, or throw. + * `observeSession` marks the run as model-facing work on that session: it is + * instrumented into the observation service (action begin/end) and tagged so + * interrupt() can kill exactly this child. Observation traffic itself never + * passes an observeSession. + */ async function runBsk( deps: ToolDeps, exec: ToolRunContext, args: string[], label: string, + observeSession?: string, ): Promise { - let result; + if (observeSession !== undefined) { + deps.observation.beginAction(observeSession, actionForLabel(label)); + } + let actionError: string | undefined; try { - result = await deps.runner.run(args, { - signal: exec.signal, - timeoutMs: deps.config.defaultTimeoutMs, - }); - } catch (error) { - if (isCommandNotFound(error)) { - throw new Error(bskInstallMessage(deps.config.bskPath)); + let result; + try { + result = await deps.runner.run(args, { + signal: exec.signal, + timeoutMs: deps.config.defaultTimeoutMs, + ...(observeSession !== undefined ? { tag: observeSession } : {}), + }); + } catch (error) { + if (isCommandNotFound(error)) { + throw new Error(bskInstallMessage(deps.config.bskPath)); + } + throw error; } + if (result.aborted) { + const error = new Error("tool call aborted"); + error.name = "AbortError"; + throw error; + } + return parseBskJson(result, label); + } catch (error) { + actionError = error instanceof Error ? error.message.split("\n")[0] : String(error); throw error; + } finally { + if (observeSession !== undefined) { + deps.observation.endAction(observeSession, actionError); + } } - if (result.aborted) { - const error = new Error("tool call aborted"); - error.name = "AbortError"; - throw error; - } - return parseBskJson(result, label); } /** Build the command line shown on the pending terminal card (pure). */ @@ -184,6 +210,7 @@ export function registerTools(deps: ToolDeps): void { browserInstanceId: reply.browser_instance_id, startedAtMs: Date.now(), }); + deps.observation.addSession(reply.session_id, args.url); try { if (args.device !== undefined) { await runBsk( @@ -191,6 +218,7 @@ export function registerTools(deps: ToolDeps): void { exec, ["emulate", "--session", reply.session_id, "--device", args.device], "emulate", + reply.session_id, ); } if (args.url !== undefined) { @@ -199,6 +227,7 @@ export function registerTools(deps: ToolDeps): void { exec, ["navigate", "--session", reply.session_id, args.url], "navigate", + reply.session_id, ); } } catch (error) { @@ -207,6 +236,7 @@ export function registerTools(deps: ToolDeps): void { .run(["session", "stop", reply.session_id], { timeoutMs: 30_000 }) .catch(() => {}); registry.remove(reply.session_id); + deps.observation.removeSession(reply.session_id); throw error; } return { @@ -250,8 +280,9 @@ export function registerTools(deps: ToolDeps): void { }, async execute(args, exec) { const sessionId = registry.resolveForStop(args.session); - await runBsk(deps, exec, ["session", "stop", sessionId], "session stop"); + await runBsk(deps, exec, ["session", "stop", sessionId], "session stop", sessionId); registry.remove(sessionId); + deps.observation.removeSession(sessionId); return { stopped: sessionId }; }, presentCall: (args) => ({ @@ -377,12 +408,13 @@ export function registerTools(deps: ToolDeps): void { if (args.waitUntil !== undefined) cmdArgs.push("--wait-until", args.waitUntil); if (args.timeoutMs !== undefined) cmdArgs.push("--timeout", `${args.timeoutMs}ms`); cmdArgs.push(args.url); - const reply = (await runBsk(deps, exec, cmdArgs, "navigate")) as { + const reply = (await runBsk(deps, exec, cmdArgs, "navigate", sessionId)) as { tab_id: number; url: string; final_url?: string; reached: string; }; + deps.observation.setUrl(sessionId, reply.final_url ?? reply.url); return { session: sessionId, tabId: reply.tab_id, @@ -448,7 +480,7 @@ export function registerTools(deps: ToolDeps): void { const cmdArgs = [kind, "--session", sessionId]; if (args.maxDepth !== undefined) cmdArgs.push("--max-depth", String(args.maxDepth)); if (args.maxTokens !== undefined) cmdArgs.push("--max-tokens", String(args.maxTokens)); - const reply = (await runBsk(deps, exec, cmdArgs, kind)) as { + const reply = (await runBsk(deps, exec, cmdArgs, kind, sessionId)) as { text: string; ref_count: number; tab_id: number; @@ -522,7 +554,7 @@ export function registerTools(deps: ToolDeps): void { if (args.button !== undefined) cmdArgs.push("--button", args.button); if (args.clickCount !== undefined) cmdArgs.push("--click-count", String(args.clickCount)); cmdArgs.push(args.target); - const reply = (await runBsk(deps, exec, cmdArgs, "click")) as { + const reply = (await runBsk(deps, exec, cmdArgs, "click", sessionId)) as { tab_id: number; x: number; y: number; @@ -580,7 +612,7 @@ export function registerTools(deps: ToolDeps): void { const cmdArgs = ["fill", "--session", sessionId, "--value", args.value]; if (args.noClear === true) cmdArgs.push("--no-clear"); cmdArgs.push(args.target); - const reply = (await runBsk(deps, exec, cmdArgs, "fill")) as { + const reply = (await runBsk(deps, exec, cmdArgs, "fill", sessionId)) as { tab_id: number; value_length: number; }; @@ -650,7 +682,7 @@ export function registerTools(deps: ToolDeps): void { } if (args.holdMs !== undefined) cmdArgs.push("--hold-ms", String(args.holdMs)); cmdArgs.push(args.key); - const reply = (await runBsk(deps, exec, cmdArgs, "press")) as { + const reply = (await runBsk(deps, exec, cmdArgs, "press", sessionId)) as { tab_id: number; key: string; }; @@ -735,7 +767,7 @@ export function registerTools(deps: ToolDeps): void { ); const cmdArgs = ["screenshot", "--session", sessionId, "--out", outPath]; if (args.ref !== undefined) cmdArgs.push("--ref", args.ref); - const reply = (await runBsk(deps, exec, cmdArgs, "screenshot")) as { + const reply = (await runBsk(deps, exec, cmdArgs, "screenshot", sessionId)) as { tab_id: number; width: number; height: number; @@ -839,7 +871,7 @@ export function registerTools(deps: ToolDeps): void { } if (args.mobile === true) cmdArgs.push("--mobile"); } - const reply = (await runBsk(deps, exec, cmdArgs, "emulate")) as { + const reply = (await runBsk(deps, exec, cmdArgs, "emulate", sessionId)) as { tab_id: number; cleared: boolean; }; diff --git a/packages/dsh-plugin-browserskill/tests/dispose.test.ts b/packages/dsh-plugin-browserskill/tests/dispose.test.ts index 0163d05..b392adb 100644 --- a/packages/dsh-plugin-browserskill/tests/dispose.test.ts +++ b/packages/dsh-plugin-browserskill/tests/dispose.test.ts @@ -58,6 +58,7 @@ describe("dispose cleanup ownership", () => { return fail(`unexpected command: ${joined}`); }, killAll() {}, + killFor: () => 0, }; const tools = new Map(); const disposers: Array<() => Promise> = []; diff --git a/packages/dsh-plugin-browserskill/tests/observation.test.ts b/packages/dsh-plugin-browserskill/tests/observation.test.ts new file mode 100644 index 0000000..b30749b --- /dev/null +++ b/packages/dsh-plugin-browserskill/tests/observation.test.ts @@ -0,0 +1,407 @@ +// ObservationService host tests: state machine, cadence/backoff, observation- +// traffic isolation, interrupt routing, the owned boundary, and the HTTP/SSE +// interface. All bsk runs are faked; the scheduler is a manual timer queue. + +import { mkdtempSync, writeFileSync } from "node:fs"; +import type { IncomingMessage, ServerResponse } from "node:http"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { describe, expect, it } from "vitest"; +import { + actionForLabel, + type ObservationEvent, + type ObservationScheduler, + ObservationService, +} from "../src/observation"; +import { registerObservationRoutes } from "../src/observation-http"; +import type { BskRunner, BskRunOptions, BskRunResult } from "../src/runner"; +import { SessionRegistry } from "../src/sessions"; + +const OPTIONS = { enabled: true, thumbnailIntervalMs: 1500, idleIntervalMs: 8000 }; + +/** Manual timer queue + controllable clock. */ +function fakeScheduler() { + let now = 1_000_000; + const timers: { handle: object; fn: () => void; ms: number }[] = []; + const scheduler: ObservationScheduler & { + advance: (ms: number) => void; + runNext: () => void; + pending: () => number[]; + } = { + now: () => now, + setTimeout: (fn, ms) => { + const handle = { fn, ms }; + timers.push({ handle, fn, ms }); + return handle; + }, + clearTimeout: (h) => { + const index = timers.findIndex((t) => t.handle === h); + if (index >= 0) timers.splice(index, 1); + }, + advance: (ms) => { + now += ms; + }, + runNext: () => { + const next = timers.shift(); + next?.fn(); + }, + pending: () => timers.map((t) => t.ms), + }; + return scheduler; +} + +interface FakeRunner extends BskRunner { + calls: { args: string[]; options: BskRunOptions }[]; + killed: string[]; +} + +/** Runner that answers screenshot with a real temp PNG and records kills. */ +function fakeRunner(opts: { screenshotFails?: boolean; killCount?: number } = {}): FakeRunner { + const calls: FakeRunner["calls"] = []; + const killed: string[] = []; + return { + calls, + killed, + async run(args: string[], options: BskRunOptions = {}): Promise { + calls.push({ args, options }); + if (args[0] === "screenshot") { + if (opts.screenshotFails) { + return { code: 1, stdout: "", stderr: "boom", timedOut: false, aborted: false }; + } + const outIndex = args.indexOf("--out") + 1; + const out = args[outIndex]; + writeFileSync(out, Buffer.from([0x89, 0x50, 0x4e, 0x47])); + return { + code: 0, + stdout: JSON.stringify({ + tab_id: 7, + width: 4, + height: 4, + format: "png", + path: out, + byte_size: 4, + }), + stderr: "", + timedOut: false, + aborted: false, + }; + } + return { code: 0, stdout: "{}", stderr: "", timedOut: false, aborted: false }; + }, + killAll() {}, + killFor(tag: string) { + killed.push(tag); + return opts.killCount ?? 1; + }, + }; +} + +function fakeCtx(attachments?: unknown) { + return { get: (key: string) => (key === "attachments" ? attachments : undefined) } as never; +} + +function setup(opts: { + runner?: FakeRunner; + attachments?: unknown; + registry?: SessionRegistry; + scheduler?: ReturnType; +}) { + const registry = opts.registry ?? new SessionRegistry(5); + const runner = opts.runner ?? fakeRunner(); + const scheduler = opts.scheduler ?? fakeScheduler(); + const attachments = + "attachments" in opts + ? opts.attachments + : { saveImage: async () => ({ attachmentId: "att-1" }) }; + const service = new ObservationService({ + ctx: fakeCtx(attachments), + runner, + registry, + options: OPTIONS, + scheduler, + }); + const events: ObservationEvent[] = []; + service.subscribe((event) => events.push(event)); + return { service, registry, runner, scheduler, events }; +} + +/** Drive the registry through a real start so the session is owned. */ +function own(registry: SessionRegistry, sessionId: string): void { + registry.reserveStart(); + registry.completeStart({ sessionId, startedAtMs: 1 }); +} + +async function flushMicrotasks(): Promise { + // setImmediate yields to the poll phase so real fs reads inside capture() settle. + for (let i = 0; i < 20; i++) await new Promise((resolve) => setImmediate(resolve)); +} + +/** Poll until the condition holds (capture completion is real async I/O). */ +async function waitFor(cond: () => boolean, timeoutMs = 2000): Promise { + const start = Date.now(); + while (!cond()) { + if (Date.now() - start > timeoutMs) throw new Error("waitFor timeout"); + await new Promise((resolve) => setImmediate(resolve)); + } +} + +describe("state machine", () => { + it("tracks add/action/end/url/remove and emits upsert/remove/reset events", () => { + const { service, events } = setup({}); + service.addSession("s1"); + expect(service.getState()).toEqual([{ sessionId: "s1", action: "idle", since: 1_000_000 }]); + service.beginAction("s1", "navigating"); + expect(service.getState()[0].action).toBe("navigating"); + service.endAction("s1"); + expect(service.getState()[0].action).toBe("idle"); + service.setUrl("s1", "https://example.com"); + expect(service.getState()[0].url).toBe("https://example.com"); + service.endAction("s1", "click failed: no ref"); + expect(service.getState()[0].lastError).toBe("click failed: no ref"); + service.removeSession("s1"); + expect(service.getState()).toEqual([]); + expect(events.map((e) => e.type)).toEqual([ + "upsert", // add + "upsert", // begin + "upsert", // end + "upsert", // setUrl + "upsert", // end with error + "remove", + ]); + service.dispose(); + expect(events[events.length - 1].type).toBe("reset"); + }); + + it("ignores instrumentation for unknown sessions (owned-only by construction)", () => { + const { service } = setup({}); + service.beginAction("foreign", "clicking"); + service.endAction("foreign"); + service.setUrl("foreign", "https://x"); + expect(service.getState()).toEqual([]); + }); +}); + +describe("thumbnail cadence", () => { + it("captures immediately on add and after action end, then fast-cadences while active", async () => { + const scheduler = fakeScheduler(); + const runner = fakeRunner(); + const { service } = setup({ runner, scheduler }); + const captures = () => runner.calls.filter((c) => c.args[0] === "screenshot").length; + service.addSession("s1"); + // add schedules an immediate (0ms) capture. + expect(scheduler.pending()).toEqual([0]); + scheduler.runNext(); + await waitFor(() => captures() === 1 && scheduler.pending().length === 1); + expect(service.getState()[0].thumbnailAttachmentId).toBe("att-1"); + // Just captured with fresh activity: next frame on the fast cadence. + expect(scheduler.pending()).toEqual([1500]); + scheduler.runNext(); + await waitFor(() => captures() === 2 && scheduler.pending().length === 1); + expect(scheduler.pending()).toEqual([1500]); + }); + + it("downclocks to the idle cadence when the session has been quiet", async () => { + const scheduler = fakeScheduler(); + const runner = fakeRunner(); + const { service, scheduler: sch } = setup({ scheduler, runner }); + const captures = () => runner.calls.filter((c) => c.args[0] === "screenshot").length; + service.addSession("s1"); + sch.runNext(); + await waitFor(() => captures() === 1 && sch.pending().length === 1); + // Fresh activity: fast cadence. + expect(sch.pending()).toEqual([1500]); + // Go quiet beyond the idle window before the next frame fires. + sch.advance(9000); + sch.runNext(); + await waitFor(() => captures() === 2 && sch.pending().length === 1); + expect(sch.pending()).toEqual([8000]); + }); + + it("backs off to the idle cadence after repeated capture failures, keeping the old frame", async () => { + const scheduler = fakeScheduler(); + const runner = fakeRunner({ screenshotFails: true }); + const { service } = setup({ runner, scheduler }); + const captures = () => runner.calls.filter((c) => c.args[0] === "screenshot").length; + service.addSession("s1"); + for (let i = 1; i <= 4; i++) { + scheduler.runNext(); + await waitFor(() => captures() === i && scheduler.pending().length === 1); + } + // 4 attempts (1 immediate + 3 fast), then backoff to the idle cadence. + expect(captures()).toBe(4); + expect(scheduler.pending()).toEqual([8000]); + expect(service.getState()[0].thumbnailAttachmentId).toBeUndefined(); + }); + + it("stays silent without an attachment store (headless composition)", async () => { + const scheduler = fakeScheduler(); + const runner = fakeRunner(); + const { service } = setup({ runner, scheduler, attachments: undefined }); + service.addSession("s1"); + scheduler.runNext(); + await flushMicrotasks(); + expect(runner.calls.filter((c) => c.args[0] === "screenshot")).toHaveLength(0); + }); +}); + +describe("observation traffic isolation", () => { + it("captures never emit action changes nor touch the registry", async () => { + const scheduler = fakeScheduler(); + const registry = new SessionRegistry(5); + own(registry, "s1"); + const { service, events } = setup({ registry, scheduler }); + service.addSession("s1"); + events.length = 0; + scheduler.runNext(); + await waitFor(() => events.filter((e) => e.type === "upsert").length === 1); + // The capture upsert carries the thumbnail but the action stays idle… + const captureEvents = events.filter((e) => e.type === "upsert"); + expect(captureEvents).toHaveLength(1); + expect(captureEvents[0].session?.action).toBe("idle"); + expect(captureEvents[0].session?.thumbnailAttachmentId).toBe("att-1"); + // …and the registry's current pointer never moved through observation. + expect(registry.current()).toBe("s1"); + expect(registry.list()[0].startedAtMs).toBe(1); + }); +}); + +describe("interrupt routing", () => { + it("interrupts the current session by default", () => { + const registry = new SessionRegistry(5); + own(registry, "s1"); + const runner = fakeRunner(); + const { service } = setup({ registry, runner }); + expect(service.interrupt()).toBe(true); + expect(runner.killed).toEqual(["s1"]); + }); + + it("interrupts the specified owned session", () => { + const registry = new SessionRegistry(5); + own(registry, "s1"); + own(registry, "s2"); + const runner = fakeRunner(); + const { service } = setup({ registry, runner }); + expect(service.interrupt("s1")).toBe(true); + expect(runner.killed).toEqual(["s1"]); + }); + + it("returns false with no current session and for foreign ids without killing", () => { + const registry = new SessionRegistry(5); + own(registry, "s1"); + const runner = fakeRunner(); + const { service } = setup({ registry, runner }); + expect(service.interrupt("foreign")).toBe(false); + registry.remove("s1"); + expect(service.interrupt()).toBe(false); + expect(runner.killed).toEqual([]); + }); +}); + +describe("HTTP/SSE interface", () => { + interface RecordedRoute { + path: string; + handler: (req: IncomingMessage, res: ServerResponse) => void | Promise; + } + + function routeHarness() { + const routes = new Map(); + const webServer = { + register(route: RecordedRoute & { kind: string }) { + routes.set(route.path, route); + return () => routes.delete(route.path); + }, + }; + return { routes, webServer }; + } + + function fakeRes() { + const chunks: string[] = []; + const res = { + status: 0, + writeHead(status: number) { + res.status = status; + return res; + }, + write(chunk: string) { + chunks.push(chunk); + }, + end(chunk?: string) { + if (chunk !== undefined) chunks.push(chunk); + }, + body: () => chunks.join(""), + }; + return { + res: res as unknown as ServerResponse & { status: number; body: () => string }, + chunks, + }; + } + + it("serves state, streams events, and routes interrupt", async () => { + const registry = new SessionRegistry(5); + own(registry, "s1"); + const runner = fakeRunner(); + const { service } = setup({ registry, runner }); + service.addSession("s1"); + + const { routes, webServer } = routeHarness(); + const ctx = { get: (key: string) => (key === "webServer" ? webServer : undefined) } as never; + const dispose = registerObservationRoutes(ctx, service); + + // state + const stateRoute = routes.get("/bsk-observation/state"); + expect(stateRoute).toBeDefined(); + const stateRes = fakeRes(); + await stateRoute?.handler({ method: "GET" } as IncomingMessage, stateRes.res as never); + expect(stateRes.res.status).toBe(200); + expect(JSON.parse(stateRes.res.body()).sessions[0].sessionId).toBe("s1"); + + // events (SSE): one write per service event + const eventsRoute = routes.get("/bsk-observation/events"); + const eventsRes = fakeRes(); + const closeHandlers: (() => void)[] = []; + const req = { + method: "GET", + on: (event: string, fn: () => void) => { + if (event === "close") closeHandlers.push(fn); + }, + } as unknown as IncomingMessage; + await eventsRoute?.handler(req, eventsRes.res as never); + service.beginAction("s1", "clicking"); + expect(eventsRes.res.body()).toContain('"action":"clicking"'); + closeHandlers.forEach((fn) => fn()); + + // interrupt (POST {sessionId}) + const interruptRoute = routes.get("/bsk-observation/interrupt"); + const interruptRes = fakeRes(); + const postReq = { + method: "POST", + on: (event: string, fn: (chunk?: string) => void) => { + if (event === "data") fn(JSON.stringify({ sessionId: "s1" })); + if (event === "end") fn(); + }, + } as unknown as IncomingMessage; + await interruptRoute?.handler(postReq, interruptRes.res as never); + expect(JSON.parse(interruptRes.res.body())).toEqual({ interrupted: true }); + + dispose(); + expect(routes.size).toBe(0); + service.dispose(); + }); + + it("registers nothing when no webServer is mounted", () => { + const { service } = setup({}); + const ctx = { get: () => undefined } as never; + const dispose = registerObservationRoutes(ctx, service); + expect(typeof dispose).toBe("function"); + dispose(); + service.dispose(); + }); +}); + +describe("actionForLabel", () => { + it("maps tool labels onto observation verbs", () => { + expect(actionForLabel("navigate")).toBe("navigating"); + expect(actionForLabel("screenshot")).toBe("capturing"); + expect(actionForLabel("session start")).toBe("starting"); + }); +}); diff --git a/packages/dsh-plugin-browserskill/tests/tools.test.ts b/packages/dsh-plugin-browserskill/tests/tools.test.ts index 8442c6f..12fcf3a 100644 --- a/packages/dsh-plugin-browserskill/tests/tools.test.ts +++ b/packages/dsh-plugin-browserskill/tests/tools.test.ts @@ -3,11 +3,29 @@ import { tmpdir } from "node:os"; import { join } from "node:path"; import type { ToolDefinition, ToolRunContext } from "@deepseek-ai/dsh-tools"; import { describe, expect, it } from "vitest"; -import type { BskRunOptions, BskRunResult } from "../src/runner"; +import { ObservationService } from "../src/observation"; +import type { BskRunner, BskRunOptions, BskRunResult } from "../src/runner"; import { SessionRegistry } from "../src/sessions"; import { type PluginConfig, registerTools } from "../src/tools"; -const CONFIG: PluginConfig = { bskPath: "bsk", defaultTimeoutMs: 120_000, maxSessions: 5 }; +const CONFIG: PluginConfig = { + bskPath: "bsk", + defaultTimeoutMs: 120_000, + maxSessions: 5, + observationEnabled: false, + thumbnailIntervalMs: 1500, + idleIntervalMs: 8000, +}; + +/** An observation service with the feature off: instrumentation becomes a no-op. */ +function disabledObservation(deps: { ctx: unknown; runner: BskRunner; registry: SessionRegistry }) { + return new ObservationService({ + ctx: deps.ctx as never, + runner: deps.runner, + registry: deps.registry, + options: { enabled: false, thumbnailIntervalMs: 1500, idleIntervalMs: 8000 }, + }); +} interface FakeCall { args: string[]; @@ -58,6 +76,7 @@ function fakeRunner(responses: Record) { }; }, killAll() {}, + killFor: () => 0, }, }; } @@ -95,7 +114,14 @@ function setup( const { ctx, tools } = makeCtx(services); const { runner, calls } = fakeRunner(responses); const registry = new SessionRegistry(maxSessions); - registerTools({ ctx: ctx as never, runner, registry, config: { ...CONFIG, maxSessions } }); + const config = { ...CONFIG, maxSessions }; + registerTools({ + ctx: ctx as never, + runner: runner as BskRunner, + registry, + config, + observation: disabledObservation({ ctx, runner: runner as BskRunner, registry }), + }); return { tools, calls, registry }; } @@ -542,8 +568,15 @@ describe("error and cancellation semantics", () => { throw Object.assign(new Error("spawn bsk ENOENT"), { code: "ENOENT" }); }, killAll() {}, + killFor: () => 0, }; - registerTools({ ctx: ctx as never, runner, registry, config: CONFIG }); + registerTools({ + ctx: ctx as never, + runner, + registry, + config: CONFIG, + observation: disabledObservation({ ctx, runner, registry }), + }); const start = tools.get("browser_session_start"); await expect(start?.execute({}, makeExec())).rejects.toThrow(/BrowserSkill must be installed/); });