diff --git a/packages/dsh-plugin-browserskill/README.md b/packages/dsh-plugin-browserskill/README.md index 53103de..872b049 100644 --- a/packages/dsh-plugin-browserskill/README.md +++ b/packages/dsh-plugin-browserskill/README.md @@ -107,6 +107,14 @@ shell's theme cannot bleed back in. the dsh `webServer` route seam (dsh 0.1's Typert Remote pipeline is closed to out-of-tree packages). All commands for one session — tool calls and frame captures alike — run through a per-session FIFO, because the daemon accepts only one unfinished command per session. +- **Trust model**: these routes expose live screenshots (and an interrupt write), so they + replicate the browser-trust fence dsh applies to its own `/api` routes: the request Host + must be a loopback authority (`localhost`, `127.0.0.0/8`, `[::1]`), a present Origin must + match the Host, `sec-fetch-site: cross-site` is refused, and POST requires an + `application/json` body (cross-site simple requests can never satisfy that). The channel is + therefore built for **loopback-only serving** — binding the dsh web server to `0.0.0.0` and + reaching it through a LAN address will (deliberately) fail the fence; do not put these + routes behind a non-loopback reverse proxy without adding your own authentication. - Configure with `observationEnabled` / `thumbnailIntervalMs` / `idleIntervalMs`. ## Behavior notes diff --git a/packages/dsh-plugin-browserskill/src/client/ObservationOverlay.tsx b/packages/dsh-plugin-browserskill/src/client/ObservationOverlay.tsx index 5959732..8c0f845 100644 --- a/packages/dsh-plugin-browserskill/src/client/ObservationOverlay.tsx +++ b/packages/dsh-plugin-browserskill/src/client/ObservationOverlay.tsx @@ -434,10 +434,15 @@ export function ObservationOverlay({ store }: { store: ObservationClientStore }) const popOut = async (): Promise => { const pip = pipApi(); if (pip === undefined) return; - const win = await pip.requestWindow({ width: size.w, height: size.h }); - cloneStylesInto(win); - win.addEventListener("pagehide", () => setPipWindow(null)); - setPipWindow(win); + try { + const win = await pip.requestWindow({ width: size.w, height: size.h }); + cloneStylesInto(win); + win.addEventListener("pagehide", () => setPipWindow(null)); + setPipWindow(win); + } catch { + // requestWindow rejects without a user gesture (or when the window was + // closed mid-request): stay on the in-page card, no state change. + } }; // Hidden while no owned session exists (and no PiP is up). @@ -496,7 +501,28 @@ export function ObservationOverlay({ store }: { store: ObservationClientStore }) data-testid="obs-resize" aria-label="Resize overlay" role="separator" + aria-valuenow={size.w} + aria-valuetext={`${Math.round(size.w)} by ${Math.round(size.h)} pixels`} + aria-valuemin={MIN_SIZE.w} + aria-valuemax={Math.round(viewport().w * 0.8)} + tabIndex={0} onPointerDown={beginDrag("resize")} + onKeyDown={(event) => { + const step = 16; + const delta = + event.key === "ArrowRight" + ? { w: step, h: 0 } + : event.key === "ArrowLeft" + ? { w: -step, h: 0 } + : event.key === "ArrowDown" + ? { w: 0, h: step } + : event.key === "ArrowUp" + ? { w: 0, h: -step } + : undefined; + if (delta === undefined) return; + event.preventDefault(); + setSize(clampSize({ w: size.w + delta.w, h: size.h + delta.h }, viewport())); + }} /> ); diff --git a/packages/dsh-plugin-browserskill/src/client/observation-store.ts b/packages/dsh-plugin-browserskill/src/client/observation-store.ts index cd40e4c..aeefa7d 100644 --- a/packages/dsh-plugin-browserskill/src/client/observation-store.ts +++ b/packages/dsh-plugin-browserskill/src/client/observation-store.ts @@ -1,14 +1,17 @@ /** * Client-side data layer for the observation overlay: initial state fetch, - * SSE increment stream, on-demand thumbnail blob loading through the - * session-authorized readAttachment path, and the interrupt call. All I/O is - * injected so tests never touch a network. + * SSE increment stream (with a state refetch on every (re)open — events that + * fired while the stream was down are otherwise lost forever), on-demand + * thumbnail blob loading through the session-authorized readAttachment path, + * and the interrupt call. All I/O is injected so tests never touch a network. */ import type { ObservationEvent, SessionObservation } from "../observation"; export interface EventSourceLike { onmessage: ((event: { data: string }) => void) | null; + /** Fires on the initial connect AND on every automatic reconnect. */ + onopen?: (() => void) | null; close(): void; } @@ -32,7 +35,8 @@ export interface ThumbnailState { export interface OverlaySnapshot { readonly sessions: readonly SessionObservation[]; - readonly connected: boolean; + /** Whether the SSE increment stream is currently subscribed. */ + readonly subscribed: boolean; readonly thumbnails: Readonly>; /** False when the host reports the browser/daemon as unreachable. */ readonly available: boolean; @@ -42,14 +46,22 @@ const STATE_URL = "/bsk-observation/state"; const EVENTS_URL = "/bsk-observation/events"; const INTERRUPT_URL = "/bsk-observation/interrupt"; +function revoke(url: string | undefined): void { + if (url !== undefined && typeof URL.revokeObjectURL === "function") { + URL.revokeObjectURL(url); + } +} + export class ObservationClientStore { private sessions = new Map(); private thumbs = new Map(); + /** sessionId → the attachment id currently displayed for it (leak guard). */ + private readonly thumbBySession = new Map(); private listeners = new Set<() => void>(); private events: EventSourceLike | undefined; private snapshot: OverlaySnapshot = { sessions: [], - connected: false, + subscribed: false, thumbnails: {}, available: true, }; @@ -68,17 +80,15 @@ export class ObservationClientStore { private publish(): void { this.snapshot = { sessions: [...this.sessions.values()], - connected: this.events !== undefined, + subscribed: this.events !== undefined, thumbnails: Object.fromEntries(this.thumbs), available: this.available, }; for (const listener of [...this.listeners]) listener(); } - /** Initial fetch + SSE subscription. Idempotent. */ - start(): void { - if (this.started) return; - this.started = true; + /** (Re)pull the full state: initial load and every SSE (re)open. */ + private refreshState(): void { void this.deps .fetchFn(STATE_URL) .then(async (res) => { @@ -87,11 +97,25 @@ export class ObservationClientStore { sessions?: SessionObservation[]; available?: boolean; }; - this.sessions = new Map((body.sessions ?? []).map((s) => [s.sessionId, s])); + const sessions = body.sessions ?? []; + this.sessions = new Map(sessions.map((s) => [s.sessionId, s])); + // Prune thumbnails of sessions that vanished while we were away. + const alive = new Set(sessions.map((s) => s.sessionId)); + for (const sessionId of [...this.thumbBySession.keys()]) { + if (!alive.has(sessionId)) this.dropThumb(sessionId); + } + for (const s of sessions) this.trackThumb(s.sessionId, s.thumbnailAttachmentId); if (typeof body.available === "boolean") this.available = body.available; this.publish(); }) .catch(() => {}); + } + + /** Initial fetch + SSE subscription. Idempotent. */ + start(): void { + if (this.started) return; + this.started = true; + this.refreshState(); const events = this.deps.eventSourceFactory(EVENTS_URL); events.onmessage = (message) => { let event: ObservationEvent; @@ -102,6 +126,10 @@ export class ObservationClientStore { } this.apply(event); }; + events.onopen = () => { + // Reconnects lose every event fired during the outage — resync. + if (this.events === events) this.refreshState(); + }; this.events = events; this.publish(); } @@ -110,23 +138,45 @@ export class ObservationClientStore { this.events?.close(); this.events = undefined; this.started = false; - for (const thumb of this.thumbs.values()) { - if (thumb.url !== undefined && typeof URL.revokeObjectURL === "function") { - URL.revokeObjectURL(thumb.url); - } - } + for (const thumb of this.thumbs.values()) revoke(thumb.url); this.thumbs.clear(); + this.thumbBySession.clear(); this.sessions.clear(); this.publish(); } + /** Forget one session's tracked thumbnail, revoking its blob URL. */ + private dropThumb(sessionId: string): void { + const attachmentId = this.thumbBySession.get(sessionId); + if (attachmentId === undefined) return; + this.thumbBySession.delete(sessionId); + revoke(this.thumbs.get(attachmentId)?.url); + this.thumbs.delete(attachmentId); + } + + /** + * Track the frame a session currently shows; replacing a frame revokes the + * old blob URL immediately — blob URLs must not accumulate one per frame. + */ + private trackThumb(sessionId: string, attachmentId: string | undefined): void { + const previous = this.thumbBySession.get(sessionId); + if (previous === attachmentId) return; + this.dropThumb(sessionId); + if (attachmentId !== undefined) this.thumbBySession.set(sessionId, attachmentId); + } + private apply(event: ObservationEvent): void { if (event.type === "reset") { this.sessions.clear(); + for (const thumb of this.thumbs.values()) revoke(thumb.url); + this.thumbs.clear(); + this.thumbBySession.clear(); } else if (event.type === "remove" && event.session !== undefined) { this.sessions.delete(event.session.sessionId); + this.dropThumb(event.session.sessionId); } else if (event.type === "upsert" && event.session !== undefined) { this.sessions.set(event.session.sessionId, event.session); + this.trackThumb(event.session.sessionId, event.session.thumbnailAttachmentId); } else if (event.type === "availability") { this.available = event.available; } @@ -144,6 +194,11 @@ export class ObservationClientStore { this.publish(); this.deps.loadImage(attachmentId).then( (url) => { + if (!this.thumbs.has(attachmentId)) { + // Replaced or removed while loading: never resurrect (or leak) it. + revoke(url); + return; + } this.thumbs.set(attachmentId, { status: "ready", url }); this.publish(); }, diff --git a/packages/dsh-plugin-browserskill/src/index.ts b/packages/dsh-plugin-browserskill/src/index.ts index e5660a0..a065aa4 100644 --- a/packages/dsh-plugin-browserskill/src/index.ts +++ b/packages/dsh-plugin-browserskill/src/index.ts @@ -90,8 +90,10 @@ export function apply( }); // Non-blocking install probe: warn early when bsk is missing instead of - // failing the first tool call with a bare spawn error. - runner.run(["status"], { timeoutMs: 10_000 }).then( + // failing the first tool call with a bare spawn error. Uses --version on + // purpose — it answers without starting the daemon (`bsk status` would + // ensure-spawn one, an expensive side effect for a probe). + runner.run(["--version"], { timeoutMs: 10_000 }).then( () => {}, (error: unknown) => { const detail = error instanceof Error ? error.message : String(error); diff --git a/packages/dsh-plugin-browserskill/src/observation-http.ts b/packages/dsh-plugin-browserskill/src/observation-http.ts index f41fef7..3225e18 100644 --- a/packages/dsh-plugin-browserskill/src/observation-http.ts +++ b/packages/dsh-plugin-browserskill/src/observation-http.ts @@ -35,6 +35,56 @@ function sendJson(res: ServerResponse, status: number, body: unknown): void { res.end(JSON.stringify(body)); } +/** + * Browser-trust fence, mirroring the one dsh applies to its /api routes (ours + * live outside that prefix, so the checks are replicated here): + * - Host must be a loopback authority (localhost / 127.0.0.0/8 / [::1]) — the + * observation channel exposes live screenshots and must never answer a LAN + * or DNS-rebound name; + * - a present Origin must be same-host with the Host header (blocks cross-site + * reads), and `sec-fetch-site: cross-site` is refused outright; + * - POST must be `application/json` — anything a cross-site *simple request* + * can send (form/plain) never reaches the handler, which kills CSRF. + */ +function fenceViolation(req: IncomingMessage): string | undefined { + const host = req.headers.host ?? ""; + const hostname = /^\[.*\](?::\d+)?$/.test(host) + ? host.slice(1, host.indexOf("]")) + : host.split(":")[0]; + const isLoopback = + hostname === "localhost" || + hostname.endsWith(".localhost") || + hostname === "::1" || + /^127\.\d{1,3}\.\d{1,3}\.\d{1,3}$/.test(hostname); + if (!isLoopback) return "host is not a loopback authority"; + const origin = req.headers.origin; + if (origin !== undefined && origin !== "null") { + let originHost: string | undefined; + try { + originHost = new URL(origin).host; + } catch { + return "unparseable Origin header"; + } + if (originHost !== host) return "Origin does not match Host"; + } + if (req.headers["sec-fetch-site"] === "cross-site") return "sec-fetch-site: cross-site"; + if (req.method === "POST") { + const contentType = req.headers["content-type"] ?? ""; + if (!/^\s*application\/json\s*(;|$)/.test(contentType)) { + return "POST requires an application/json body"; + } + } + return undefined; +} + +/** Run the fence; returns true when the request was rejected (handled). */ +function fenceRejected(req: IncomingMessage, res: ServerResponse): boolean { + const violation = fenceViolation(req); + if (violation === undefined) return false; + sendJson(res, 403, { error: `forbidden: ${violation}` }); + return true; +} + /** * Register the observation routes. No-op (with a console note) when the * composition has no web server. @@ -57,6 +107,7 @@ export function registerObservationRoutes( sendJson(res, 405, { error: "method not allowed" }); return; } + if (fenceRejected(req, res)) return; sendJson(res, 200, { sessions: observation.getState(), available: observation.isAvailable(), @@ -71,6 +122,7 @@ export function registerObservationRoutes( sendJson(res, 405, { error: "method not allowed" }); return; } + if (fenceRejected(req, res)) return; res.writeHead(200, { "content-type": "text/event-stream", "cache-control": "no-cache", @@ -80,7 +132,7 @@ export function registerObservationRoutes( res.write(`data: ${JSON.stringify(event)}\n\n`); }); const heartbeat = setInterval(() => res.write(": heartbeat\n\n"), SSE_HEARTBEAT_MS); - req.on("close", () => { + res.on("close", () => { clearInterval(heartbeat); unsubscribe(); }); @@ -94,6 +146,7 @@ export function registerObservationRoutes( sendJson(res, 405, { error: "method not allowed" }); return; } + if (fenceRejected(req, res)) return; let body = ""; req.on("data", (chunk: Buffer | string) => { body += chunk; @@ -121,6 +174,7 @@ export function registerObservationRoutes( sendJson(res, 405, { error: "method not allowed" }); return; } + if (fenceRejected(req, res)) return; const pathname = decodeURIComponent(new URL(req.url ?? "/", "http://x").pathname); const attachmentId = pathname.slice(`${ROUTE_BASE}/thumbnail/`.length); const frame = await observation.readThumbnail(attachmentId); diff --git a/packages/dsh-plugin-browserskill/src/observation.ts b/packages/dsh-plugin-browserskill/src/observation.ts index ba5173a..afc761a 100644 --- a/packages/dsh-plugin-browserskill/src/observation.ts +++ b/packages/dsh-plugin-browserskill/src/observation.ts @@ -11,7 +11,7 @@ * events, and never move the registry's current pointer. */ -import { readFile } from "node:fs/promises"; +import { readFile, unlink } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; import type { Context } from "@deepseek-ai/cordis"; @@ -179,6 +179,10 @@ export class ObservationService { this.cancelCapture(sessionId); this.captureFailures.delete(sessionId); this.lastActivity.delete(sessionId); + const entry = this.observations.get(sessionId); + if (entry?.thumbnailAttachmentId !== undefined) { + this.thumbRefs.delete(entry.thumbnailAttachmentId); + } if (this.observations.delete(sessionId)) { this.emit({ type: "remove", @@ -324,8 +328,11 @@ export class ObservationService { return; } this.captureInFlight.add(sessionId); + // One fixed scratch path per session (overwritten each frame) and always + // unlinked after the bytes are read — /tmp must not grow without bound. + const outPath = join(tmpdir(), `bsk-obs-${sessionId}.png`); + let writtenPath = outPath; try { - const outPath = join(tmpdir(), `bsk-obs-${sessionId}-${this.scheduler.now()}.png`); const result = await this.deps.queue.run(sessionId, () => this.deps.runner.run(["screenshot", "--session", sessionId, "--out", outPath], { timeoutMs: 15_000, @@ -346,7 +353,8 @@ export class ObservationService { throw new Error(`screenshot exited ${result.code}`); } const reply = JSON.parse(result.stdout) as { path?: string }; - const data = await readFile(reply.path ?? outPath); + writtenPath = reply.path ?? outPath; + const data = await readFile(writtenPath); const ref = await attachments.saveImage({ data, mediaType: "image/png", @@ -357,6 +365,11 @@ export class ObservationService { this.captureFailures.delete(sessionId); this.globalFailures = 0; this.setAvailable(true); + // Drop the replaced frame's reference — the store keeps the bytes, but + // this map (and the old id's read path) must not grow without bound. + if (entry.thumbnailAttachmentId !== undefined) { + this.thumbRefs.delete(entry.thumbnailAttachmentId); + } this.thumbRefs.set(String(ref.attachmentId), ref); this.put({ ...entry, thumbnailAttachmentId: String(ref.attachmentId) }); } catch { @@ -365,6 +378,7 @@ export class ObservationService { this.globalFailures += 1; if (this.globalFailures >= FAILURE_BACKOFF_THRESHOLD) this.setAvailable(false); } finally { + await unlink(writtenPath).catch(() => {}); this.captureInFlight.delete(sessionId); if (!this.disposed && this.observations.has(sessionId)) this.scheduleCapture(sessionId); } diff --git a/packages/dsh-plugin-browserskill/src/queue.ts b/packages/dsh-plugin-browserskill/src/queue.ts index 2e53204..743096e 100644 --- a/packages/dsh-plugin-browserskill/src/queue.ts +++ b/packages/dsh-plugin-browserskill/src/queue.ts @@ -43,14 +43,15 @@ export class KeyedExecutor { ); }); }); - // The stored tail never rejects, so one failure never strands the queue. - this.tails.set( - key, - task.then( - () => {}, - () => {}, - ), - ); + // The stored tail never rejects, so one failure never strands the queue; + // and it chains BOTH the previous tail and this task, so a task aborted + // while still queued cannot let a later task overlap the one before it. + const tail = Promise.allSettled([previous, task]).then(() => {}); + this.tails.set(key, tail); + // Drop the entry once the queue drains, so the map tracks live keys only. + void tail.then(() => { + if (this.tails.get(key) === tail) this.tails.delete(key); + }); return task; } } diff --git a/packages/dsh-plugin-browserskill/src/runner.ts b/packages/dsh-plugin-browserskill/src/runner.ts index 023e2f0..6505a47 100644 --- a/packages/dsh-plugin-browserskill/src/runner.ts +++ b/packages/dsh-plugin-browserskill/src/runner.ts @@ -48,18 +48,13 @@ export interface BskRunResult { export interface BskRunOptions { signal?: AbortSignal; timeoutMs?: number; + /** Opaque routing tag (e.g. a session id) enabling per-tag kills. */ + tag?: string; } /** 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; diff --git a/packages/dsh-plugin-browserskill/src/tools.ts b/packages/dsh-plugin-browserskill/src/tools.ts index c4b8c46..77dfa63 100644 --- a/packages/dsh-plugin-browserskill/src/tools.ts +++ b/packages/dsh-plugin-browserskill/src/tools.ts @@ -6,7 +6,7 @@ * the completed card carries the output. */ -import { readFile } from "node:fs/promises"; +import { readFile, unlink } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; import type { Context } from "@deepseek-ai/cordis"; @@ -70,19 +70,25 @@ async function runBsk( label: string, observeSession?: string, ): Promise { - if (observeSession !== undefined) { - deps.observation.beginAction(observeSession, actionForLabel(label)); - } + // beginAction must fire only when the task actually starts inside the + // per-session queue — marking it while still queued would overwrite the + // in-flight action's label (and flash it idle on a queued abort). + let began = false; let actionError: string | undefined; try { let result; try { - const runOnce = () => - deps.runner.run(args, { + const runOnce = () => { + if (observeSession !== undefined) { + began = true; + deps.observation.beginAction(observeSession, actionForLabel(label)); + } + return deps.runner.run(args, { signal: exec.signal, timeoutMs: deps.config.defaultTimeoutMs, ...(observeSession !== undefined ? { tag: observeSession } : {}), }); + }; result = observeSession !== undefined ? await deps.queue.run(observeSession, runOnce, exec.signal) @@ -103,7 +109,7 @@ async function runBsk( actionError = error instanceof Error ? error.message.split("\n")[0] : String(error); throw error; } finally { - if (observeSession !== undefined) { + if (observeSession !== undefined && began) { deps.observation.endAction(observeSession, actionError); } } @@ -240,9 +246,20 @@ export function registerTools(deps: ToolDeps): void { } } catch (error) { // A half-initialized session must not leak: stop it before surfacing. - await deps.runner - .run(["session", "stop", reply.session_id], { timeoutMs: 30_000 }) - .catch(() => {}); + // The stop funnels through the session's queue (an in-flight command + // — e.g. an observation capture — would otherwise make the daemon + // refuse it), with one retry for good measure. + for (let attempt = 0; attempt < 2; attempt++) { + try { + const stopped = await deps.queue.run(reply.session_id, () => + deps.runner.run(["session", "stop", reply.session_id], { timeoutMs: 30_000 }), + ); + if (stopped.code === 0) break; + } catch { + // retry once, then give up quietly (registry/observation are + // cleaned below either way; the daemon reaps orphans). + } + } registry.remove(reply.session_id); deps.observation.removeSession(reply.session_id); throw error; @@ -775,36 +792,47 @@ 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", sessionId)) as { - tab_id: number; - width: number; - height: number; - path: string; - byte_size: number; - }; - const data = await readFile(reply.path); - if (exec.signal.aborted) throw abortError(); - const ref = await trySaveScreenshot(deps.ctx, exec, data, `screenshot-${sessionId}.png`); - return { - session: sessionId, - tabId: reply.tab_id, - path: reply.path, - width: reply.width, - height: reply.height, - byteSize: reply.byte_size, - ...(ref !== undefined - ? { - image: { - attachmentId: String(ref.attachmentId), - mediaType: "image/png" as const, - bytes: ref.bytes, - width: ref.width, - height: ref.height, - ...(ref.name !== undefined ? { name: ref.name } : {}), - }, - } - : {}), - }; + // The scratch PNG is only kept when it IS the model-facing artifact + // (no attachment store to inline through); once the bytes are saved + // into the store — and on every failure/abort path — it is deleted. + let keepFile = false; + let writtenPath: string | undefined; + try { + const reply = (await runBsk(deps, exec, cmdArgs, "screenshot", sessionId)) as { + tab_id: number; + width: number; + height: number; + path: string; + byte_size: number; + }; + writtenPath = reply.path; + const data = await readFile(reply.path); + if (exec.signal.aborted) throw abortError(); + const ref = await trySaveScreenshot(deps.ctx, exec, data, `screenshot-${sessionId}.png`); + keepFile = ref === undefined; + return { + session: sessionId, + tabId: reply.tab_id, + path: reply.path, + width: reply.width, + height: reply.height, + byteSize: reply.byte_size, + ...(ref !== undefined + ? { + image: { + attachmentId: String(ref.attachmentId), + mediaType: "image/png" as const, + bytes: ref.bytes, + width: ref.width, + height: ref.height, + ...(ref.name !== undefined ? { name: ref.name } : {}), + }, + } + : {}), + }; + } finally { + if (!keepFile) await unlink(writtenPath ?? outPath).catch(() => {}); + } }, presentCall: (args) => ({ card: "terminal", @@ -835,7 +863,12 @@ export function registerTools(deps: ToolDeps): void { }, width: { type: "integer", description: "Viewport width in CSS pixels (requires height)." }, height: { type: "integer", description: "Viewport height in CSS pixels (requires width)." }, - mobile: { type: "boolean", description: "Emulate a mobile viewport." }, + mobile: { + type: "boolean", + description: + "Emulate a mobile viewport. Requires width+height — the bsk daemon refuses " + + "--mobile without viewport dimensions, so this flag cannot be used alone.", + }, off: { type: "boolean", description: "Clear every emulation override on the tab." }, }, output: { @@ -864,7 +897,9 @@ export function registerTools(deps: ToolDeps): void { throw new Error("off is mutually exclusive with device/width/height"); } } else if (args.device === undefined && args.width === undefined) { - throw new Error("nothing to apply: pass device, width+height, or off"); + throw new Error( + "nothing to apply: pass device or width+height (mobile also requires width+height), or off", + ); } if ((args.width === undefined) !== (args.height === undefined)) { throw new Error("width and height must be given together"); diff --git a/packages/dsh-plugin-browserskill/tests/client/observation-store.test.ts b/packages/dsh-plugin-browserskill/tests/client/observation-store.test.ts index 41fbc84..b0c7714 100644 --- a/packages/dsh-plugin-browserskill/tests/client/observation-store.test.ts +++ b/packages/dsh-plugin-browserskill/tests/client/observation-store.test.ts @@ -111,3 +111,87 @@ describe("ObservationClientStore", () => { expect(store.getSnapshot().sessions).toHaveLength(0); }); }); + +describe("thumbnail blob URL lifecycle", () => { + function withRevokeSpy() { + const revoke = vi.fn(); + (URL as unknown as { revokeObjectURL: unknown }).revokeObjectURL = revoke; + return revoke; + } + + it("replacing a frame revokes the old blob URL and drops its entry", async () => { + const revoke = withRevokeSpy(); + const { store, esInstances } = harness([{ ...OBS_IDLE, thumbnailAttachmentId: "a1" }]); + store.start(); + await vi.waitFor(() => expect(store.getSnapshot().sessions).toHaveLength(1)); + store.ensureThumbnail("a1"); + await vi.waitFor(() => + expect(store.getSnapshot().thumbnails["a1"]).toEqual({ status: "ready", url: "blob:url-a1" }), + ); + emit(esInstances[0], { + type: "upsert", + session: { ...OBS_IDLE, thumbnailAttachmentId: "a2" }, + }); + expect(store.getSnapshot().thumbnails["a1"]).toBeUndefined(); + expect(revoke).toHaveBeenCalledWith("blob:url-a1"); + store.stop(); + }); + + it("remove and reset revoke the session's tracked frame", async () => { + const revoke = withRevokeSpy(); + const { store, esInstances } = harness([{ ...OBS_IDLE, thumbnailAttachmentId: "a1" }]); + store.start(); + await vi.waitFor(() => expect(store.getSnapshot().sessions).toHaveLength(1)); + store.ensureThumbnail("a1"); + await vi.waitFor(() => expect(store.getSnapshot().thumbnails["a1"].status).toBe("ready")); + emit(esInstances[0], { + type: "remove", + session: { sessionId: "s1", action: "idle", since: 0 }, + }); + expect(revoke).toHaveBeenCalledWith("blob:url-a1"); + expect(store.getSnapshot().thumbnails["a1"]).toBeUndefined(); + store.stop(); + }); + + it("a load that finishes after replacement never resurrects the old frame", async () => { + const revoke = withRevokeSpy(); + const { store, esInstances, loadImage } = harness([ + { ...OBS_IDLE, thumbnailAttachmentId: "a1" }, + ]); + let resolveA1!: (url: string) => void; + loadImage.mockImplementation((id: string) => + id === "a1" + ? new Promise((resolve) => { + resolveA1 = resolve; + }) + : Promise.resolve(`blob:url-${id}`), + ); + store.start(); + await vi.waitFor(() => expect(store.getSnapshot().sessions).toHaveLength(1)); + store.ensureThumbnail("a1"); + // the frame is replaced while a1 is still loading + emit(esInstances[0], { + type: "upsert", + session: { ...OBS_IDLE, thumbnailAttachmentId: "a2" }, + }); + resolveA1("blob:url-a1-late"); + await new Promise((resolve) => setImmediate(resolve)); + expect(store.getSnapshot().thumbnails["a1"]).toBeUndefined(); + expect(revoke).toHaveBeenCalledWith("blob:url-a1-late"); + store.stop(); + }); +}); + +describe("SSE (re)open resync", () => { + it("refetches the full state when the stream (re)opens", async () => { + const { store, fetches, esInstances } = harness([OBS_IDLE]); + store.start(); + await vi.waitFor(() => expect(store.getSnapshot().sessions).toHaveLength(1)); + const before = fetches.filter((f) => f.url === "/bsk-observation/state").length; + esInstances[0].onopen?.(); + await vi.waitFor(() => + expect(fetches.filter((f) => f.url === "/bsk-observation/state").length).toBe(before + 1), + ); + store.stop(); + }); +}); diff --git a/packages/dsh-plugin-browserskill/tests/observation.test.ts b/packages/dsh-plugin-browserskill/tests/observation.test.ts index 87a3137..ce6f79b 100644 --- a/packages/dsh-plugin-browserskill/tests/observation.test.ts +++ b/packages/dsh-plugin-browserskill/tests/observation.test.ts @@ -2,7 +2,7 @@ // 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 { existsSync, mkdtempSync, writeFileSync } from "node:fs"; import type { IncomingMessage, ServerResponse } from "node:http"; import { tmpdir } from "node:os"; import { join } from "node:path"; @@ -344,6 +344,7 @@ describe("HTTP/SSE interface", () => { function fakeRes() { const chunks: string[] = []; + const closeHandlers: (() => void)[] = []; const res = { status: 0, writeHead(status: number) { @@ -356,14 +357,35 @@ describe("HTTP/SSE interface", () => { end(chunk?: string) { if (chunk !== undefined) chunks.push(chunk); }, + on(event: string, fn: () => void) { + if (event === "close") closeHandlers.push(fn); + }, + close: () => closeHandlers.forEach((fn) => fn()), body: () => chunks.join(""), }; return { - res: res as unknown as ServerResponse & { status: number; body: () => string }, + res: res as unknown as ServerResponse & { + status: number; + body: () => string; + close: () => void; + }, chunks, }; } + /** Loopback GET/POST request stubs the fence accepts. */ + function fakeReq(overrides: Record = {}): IncomingMessage { + return { + method: "GET", + headers: { host: "127.0.0.1:3999" }, + on: (event: string, fn: (chunk?: string) => void) => { + if (event === "data") fn(JSON.stringify({ sessionId: "s1" })); + if (event === "end") fn(); + }, + ...overrides, + } as unknown as IncomingMessage; + } + it("serves state, streams events, and routes interrupt", async () => { const registry = new SessionRegistry(5); own(registry, "s1"); @@ -379,35 +401,25 @@ describe("HTTP/SSE interface", () => { const stateRoute = routes.get("/bsk-observation/state"); expect(stateRoute).toBeDefined(); const stateRes = fakeRes(); - await stateRoute?.handler({ method: "GET" } as IncomingMessage, stateRes.res as never); + await stateRoute?.handler(fakeReq(), 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); + await eventsRoute?.handler(fakeReq(), eventsRes.res as never); service.beginAction("s1", "clicking"); expect(eventsRes.res.body()).toContain('"action":"clicking"'); - closeHandlers.forEach((fn) => fn()); + eventsRes.res.close(); // interrupt (POST {sessionId}) const interruptRoute = routes.get("/bsk-observation/interrupt"); const interruptRes = fakeRes(); - const postReq = { + const postReq = fakeReq({ 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; + headers: { host: "127.0.0.1:3999", "content-type": "application/json" }, + }); await interruptRoute?.handler(postReq, interruptRes.res as never); expect(JSON.parse(interruptRes.res.body())).toEqual({ interrupted: true }); @@ -416,6 +428,57 @@ describe("HTTP/SSE interface", () => { service.dispose(); }); + it("fences off non-loopback, cross-site, and simple-request traffic", async () => { + const registry = new SessionRegistry(5); + own(registry, "s1"); + const { service } = setup({ registry }); + service.addSession("s1"); + const { routes, webServer } = routeHarness(); + const ctx = { get: (key: string) => (key === "webServer" ? webServer : undefined) } as never; + const dispose = registerObservationRoutes(ctx, service); + const stateRoute = routes.get("/bsk-observation/state"); + const interruptRoute = routes.get("/bsk-observation/interrupt"); + + const getState = async (headers: Record) => { + const res = fakeRes(); + await stateRoute?.handler(fakeReq({ headers }), res.res as never); + return res.res.status; + }; + + // Host must be a loopback authority. + expect(await getState({ host: "192.168.1.20:3999" })).toBe(403); + expect(await getState({ host: "attacker.example" })).toBe(403); + expect(await getState({ host: "localhost:3999" })).toBe(200); + expect(await getState({ host: "[::1]:3999" })).toBe(200); + // A present Origin must match the Host (DNS-rebinding reads die here). + expect(await getState({ host: "127.0.0.1:3999", origin: "http://evil.example" })).toBe(403); + expect(await getState({ host: "127.0.0.1:3999", origin: "http://127.0.0.1:3999" })).toBe(200); + // Explicitly cross-site fetches are refused even without an Origin. + expect(await getState({ host: "127.0.0.1:3999", "sec-fetch-site": "cross-site" })).toBe(403); + expect(await getState({ host: "127.0.0.1:3999", "sec-fetch-site": "same-origin" })).toBe(200); + + // POST must be application/json — a cross-site simple request can never comply. + const post = async (contentType: string | undefined) => { + const res = fakeRes(); + const req = fakeReq({ + method: "POST", + headers: { + host: "127.0.0.1:3999", + ...(contentType !== undefined ? { "content-type": contentType } : {}), + }, + }); + await interruptRoute?.handler(req, res.res as never); + return res.res.status; + }; + expect(await post("text/plain")).toBe(403); + expect(await post("application/x-www-form-urlencoded")).toBe(403); + expect(await post(undefined)).toBe(403); + expect(await post("application/json; charset=utf-8")).toBe(200); + + dispose(); + service.dispose(); + }); + it("registers nothing when no webServer is mounted", () => { const { service } = setup({}); const ctx = { get: () => undefined } as never; @@ -434,7 +497,12 @@ describe("availability and dead sessions", () => { service.addSession("s1"); for (let i = 1; i <= 3; i++) { scheduler.runNext(); - await waitFor(() => runner.calls.filter((c) => c.args[0] === "screenshot").length === i); + // the frame's scratch-file unlink lands before the next timer is queued + await waitFor( + () => + runner.calls.filter((c) => c.args[0] === "screenshot").length === i && + scheduler.pending().length === 1, + ); } expect(service.isAvailable()).toBe(false); expect(events.some((e) => e.type === "availability" && e.available === false)).toBe(true); @@ -498,3 +566,79 @@ describe("actionForLabel", () => { expect(actionForLabel("session start")).toBe("starting"); }); }); + +describe("scratch files and thumbnail references", () => { + it("reuses one scratch path per session and deletes it after reading", async () => { + const scheduler = fakeScheduler(); + const runner = fakeRunner(); + const { service } = setup({ runner, scheduler }); + service.addSession("s1"); + // first frame (scheduled with delay 0) + scheduler.runNext(); + await waitFor( + () => + service.getState()[0]?.thumbnailAttachmentId !== undefined && + scheduler.pending().length === 1, + ); + const firstOut = runner.calls.filter((c) => c.args[0] === "screenshot").at(-1)?.args; + const pathOf = (args: string[]) => args[args.indexOf("--out") + 1]; + expect(pathOf(firstOut ?? [])).toBe(join(tmpdir(), "bsk-obs-s1.png")); + expect(existsSync(pathOf(firstOut ?? []))).toBe(false); + // second frame: same path, deleted again + service.endAction("s1"); + scheduler.runNext(); + await waitFor( + () => + runner.calls.filter((c) => c.args[0] === "screenshot").length >= 2 && + scheduler.pending().length === 1, + ); + const secondOut = runner.calls.filter((c) => c.args[0] === "screenshot").at(-1)?.args; + expect(pathOf(secondOut ?? [])).toBe(join(tmpdir(), "bsk-obs-s1.png")); + await waitFor(() => !existsSync(pathOf(secondOut ?? []))); + service.dispose(); + }); + + it("drops the replaced frame reference and clears it on remove", async () => { + let counter = 0; + const attachments = { + saveImage: async (input: { data: Uint8Array; name?: string }) => { + counter += 1; + return { + attachmentId: `att-${counter}`, + mediaType: "image/png", + bytes: input.data.byteLength, + width: 4, + height: 4, + name: input.name, + }; + }, + readImage: async (ref: { attachmentId: unknown }) => ({ + data: new Uint8Array([1, 2, 3]), + attachment: ref, + }), + }; + const scheduler = fakeScheduler(); + const { service } = setup({ scheduler, attachments }); + service.addSession("s1"); + scheduler.runNext(); + await waitFor( + () => + service.getState()[0]?.thumbnailAttachmentId === "att-1" && + scheduler.pending().length === 1, + ); + service.endAction("s1"); + scheduler.runNext(); + await waitFor( + () => + service.getState()[0]?.thumbnailAttachmentId === "att-2" && + scheduler.pending().length === 1, + ); + // the replaced frame's read path is gone + expect(await service.readThumbnail("att-1")).toBeUndefined(); + expect((await service.readThumbnail("att-2"))?.mediaType).toBe("image/png"); + // removing the session clears its live reference too + service.removeSession("s1"); + expect(await service.readThumbnail("att-2")).toBeUndefined(); + service.dispose(); + }); +}); diff --git a/packages/dsh-plugin-browserskill/tests/queue.test.ts b/packages/dsh-plugin-browserskill/tests/queue.test.ts new file mode 100644 index 0000000..b1b8b03 --- /dev/null +++ b/packages/dsh-plugin-browserskill/tests/queue.test.ts @@ -0,0 +1,85 @@ +// KeyedExecutor: per-key FIFO guarantees, including the abort-while-queued +// race that used to release the tail early (letting a later task overlap the +// one still running on the same session). + +import { describe, expect, it } from "vitest"; +import { KeyedExecutor } from "../src/queue"; + +describe("KeyedExecutor", () => { + it("runs same-key tasks in FIFO order", async () => { + const queue = new KeyedExecutor(); + const order: string[] = []; + const gate = new Promise((resolve) => setTimeout(resolve, 5)); + await Promise.all([ + queue.run("k", async () => { + await gate; + order.push("a"); + }), + queue.run("k", async () => { + order.push("b"); + }), + queue.run("k", async () => { + order.push("c"); + }), + ]); + expect(order).toEqual(["a", "b", "c"]); + }); + + it("a task aborted while queued does not let the next one overlap the runner", async () => { + const queue = new KeyedExecutor(); + let releaseA!: () => void; + const aRuns = new Promise((resolve) => { + releaseA = resolve; + }); + const events: string[] = []; + + // A holds the session until releaseA. + const a = queue.run("s1", async () => { + events.push("a:start"); + await aRuns; + events.push("a:end"); + }); + // B is queued behind A and aborted before A finishes. + const abortB = new AbortController(); + const b = queue.run( + "s1", + async () => { + events.push("b:start"); + }, + abortB.signal, + ); + // C is queued behind B. + const c = queue.run("s1", async () => { + events.push("c:start"); + }); + + abortB.abort(); + await expect(b).rejects.toMatchObject({ name: "AbortError" }); + // Give C every chance to (wrongly) start while A is still in flight. + await new Promise((resolve) => setTimeout(resolve, 20)); + expect(events).toEqual(["a:start"]); + + releaseA(); + await a; + await c; + expect(events).toEqual(["a:start", "a:end", "c:start"]); + }); + + it("a failed task never strands the queue", async () => { + const queue = new KeyedExecutor(); + await expect( + queue.run("k", async () => { + throw new Error("boom"); + }), + ).rejects.toThrow("boom"); + await expect(queue.run("k", async () => "ok")).resolves.toBe("ok"); + }); + + it("drops the tail entry once the queue drains", async () => { + const queue = new KeyedExecutor(); + const internals = queue as unknown as { tails: Map> }; + await queue.run("k", async () => {}); + await new Promise((resolve) => setImmediate(resolve)); + expect(internals.tails.size).toBe(0); + }); +}); diff --git a/packages/dsh-plugin-browserskill/tests/tools.test.ts b/packages/dsh-plugin-browserskill/tests/tools.test.ts index 688bf4a..e62cde6 100644 --- a/packages/dsh-plugin-browserskill/tests/tools.test.ts +++ b/packages/dsh-plugin-browserskill/tests/tools.test.ts @@ -1,8 +1,8 @@ -import { mkdtempSync, writeFileSync } from "node:fs"; +import { existsSync, mkdtempSync, writeFileSync } from "node:fs"; 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 { describe, expect, it, vi } from "vitest"; import { ObservationService } from "../src/observation"; import { KeyedExecutor } from "../src/queue"; import type { BskRunner, BskRunOptions, BskRunResult } from "../src/runner"; @@ -621,3 +621,180 @@ describe("presentation", () => { expect(view).toMatchObject({ card: "terminal", output: "[session s1] screenshot of tab 7" }); }); }); + +describe("screenshot scratch file lifecycle", () => { + /** Runner that honors --out like the real bsk (writes the PNG there). */ + function outWritingRunner() { + const calls: FakeCall[] = []; + return { + calls, + runner: { + async run(args: string[], options: BskRunOptions = {}): Promise { + calls.push({ args, options }); + if (args[0] === "session") { + return { + code: 0, + stdout: JSON.stringify({ session_id: "s1", browser_instance_id: "chrome-1" }), + stderr: "", + timedOut: false, + aborted: false, + }; + } + if (args[0] === "screenshot") { + const out = args[args.indexOf("--out") + 1]; + 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: 2, stdout: "{}", stderr: "", timedOut: false, aborted: false }; + }, + killAll() {}, + killFor: () => 0, + } as BskRunner, + }; + } + + function setupWithRunner(runner: BskRunner, services: Record = {}) { + const { ctx, tools } = makeCtx(services); + const registry = new SessionRegistry(5); + registerTools({ + ctx: ctx as never, + runner, + registry, + config: CONFIG, + observation: disabledObservation({ ctx, runner, registry }), + queue: new KeyedExecutor(), + }); + return { tools, registry }; + } + + const IMAGE_SERVICES = { + attachments: { + imageLimits: { mediaTypes: ["image/png"], maxImageBytes: 1e7, maxMessageImageBytes: 1e7 }, + saveImage: async (input: { data: Uint8Array; mediaType: string }) => ({ + attachmentId: "att-1", + mediaType: input.mediaType, + bytes: input.data.byteLength, + width: 4, + height: 4, + }), + }, + llm: { resolveModelInfo: async () => ({ inputModalities: ["text", "image"] }) }, + }; + + it("deletes the scratch PNG once the bytes are in the attachment store", async () => { + const { runner, calls } = outWritingRunner(); + const { tools } = setupWithRunner(runner, IMAGE_SERVICES); + await startSession(tools); + const exec = makeExec({ + agent: { + session: { requestHeader: () => ({ config: { provider: "p", model: "vl" } }) }, + options: {}, + }, + }); + const screenshot = tools.get("browser_screenshot"); + const value = (await screenshot?.execute({}, exec)) as { image?: unknown; path: string }; + expect(value.image).toBeDefined(); + const out = calls.find((c) => c.args[0] === "screenshot")?.args.at(-1); + expect(out).toBeDefined(); + expect(existsSync(out ?? "")).toBe(false); + }); + + it("keeps the PNG only when it is the model-facing artifact (no store)", async () => { + const { runner, calls } = outWritingRunner(); + const { tools } = setupWithRunner(runner); + await startSession(tools); + const screenshot = tools.get("browser_screenshot"); + const value = (await screenshot?.execute({}, makeExec())) as { image?: unknown; path: string }; + expect(value.image).toBeUndefined(); + const out = calls.find((c) => c.args[0] === "screenshot")?.args.at(-1); + expect(existsSync(out ?? "")).toBe(true); + }); +}); + +describe("observation action instrumentation timing", () => { + it("a queued call never overwrites the in-flight action label", async () => { + const { ctx, tools } = makeCtx(); + const registry = new SessionRegistry(5); + let releaseNavigate!: () => void; + const navigateGate = new Promise((resolve) => { + releaseNavigate = resolve; + }); + const runner: BskRunner = { + async run(args: string[]): Promise { + if (args[0] === "session") { + return { + code: 0, + stdout: JSON.stringify({ session_id: "s1", browser_instance_id: "chrome-1" }), + stderr: "", + timedOut: false, + aborted: false, + }; + } + if (args[0] === "navigate") { + await navigateGate; + return { + code: 0, + stdout: JSON.stringify({ tab_id: 7, url: "http://x" }), + stderr: "", + timedOut: false, + aborted: false, + }; + } + return { + code: 0, + stdout: JSON.stringify({ text: "ok", ref_count: 0, tab_id: 7, truncated: false }), + stderr: "", + timedOut: false, + aborted: false, + }; + }, + killAll() {}, + killFor: () => 0, + }; + const queue = new KeyedExecutor(); + const observation = new ObservationService({ + ctx: ctx as never, + runner, + registry, + queue, + options: { enabled: true, thumbnailIntervalMs: 1500, idleIntervalMs: 8000 }, + }); + registerTools({ ctx: ctx as never, runner, registry, config: CONFIG, observation, queue }); + + await startSession(tools); + const navigate = tools.get("browser_navigate"); + const snapshot = tools.get("browser_snapshot"); + const a = navigate?.execute({ url: "http://x" }, makeExec()) as Promise; + // A starts immediately (empty queue) and blocks on the gate. + await vi.waitFor(() => + expect(observation.getState().find((s) => s.sessionId === "s1")?.action).toBe("navigating"), + ); + // B queues behind A: the label must stay "navigating" until A settles. + const b = snapshot?.execute({}, makeExec()) as Promise; + await new Promise((resolve) => setTimeout(resolve, 30)); + expect(observation.getState().find((s) => s.sessionId === "s1")?.action).toBe("navigating"); + releaseNavigate(); + await a; + await vi.waitFor(() => + expect(observation.getState().find((s) => s.sessionId === "s1")?.action).toBe("snapshotting"), + ); + await b; + await vi.waitFor(() => + expect(observation.getState().find((s) => s.sessionId === "s1")?.action).toBe("idle"), + ); + observation.dispose(); + }); +});