fix(dsh-plugin-browserskill): address review findings (leaks, trust fence, queue race)

Blocking:
- B1: delete screenshot scratch PNGs — observation frames reuse one fixed
  per-session path and unlink after every read (finally); browser_screenshot
  unlinks once the bytes are in the attachment store (kept only when the file
  itself is the model-facing artifact).
- B2: replicate dsh's browser-trust fence on /bsk-observation/*: loopback
  Host only, Origin must match Host, sec-fetch-site: cross-site refused, POST
  requires application/json; README documents the loopback trust premise and
  the 0.0.0.0 warning.

Major:
- M1: KeyedExecutor tail chains previous+task (allSettled) so a task aborted
  while queued cannot release the next one into the running session.
- M2: client tracks the live frame per session — replacing/removing/reset
  revokes the old blob URL at once; loads settling after replacement never
  resurrect it.

Minor: drop replaced thumbRefs (+on remove); beginAction fires inside the
queue (no label overwrite while queued, no idle flash); half-initialized
session cleanup stops through the queue with one retry; SSE (re)open refetches
/state (+ snapshot flag renamed subscribed); install probe uses --version (no
daemon spawn); emulate mobile documents the width+height requirement (the
daemon refuses it alone — verified).

Nit: single BskRunOptions declaration; tails map entry dropped on drain; SSE
cleanup on res close; popOut catches requestWindow rejections; resize handle
gains keyboard control (arrows, 16px steps, aria value attrs).

Tests: +14 (fence rules, scratch lifecycle, queue race, blob revocation,
instrumentation timing, SSE resync). 113/113 green.
This commit is contained in:
BB-fat
2026-08-14 11:56:02 +00:00
parent 52aa356153
commit a4e668542d
13 changed files with 784 additions and 104 deletions
@@ -434,10 +434,15 @@ export function ObservationOverlay({ store }: { store: ObservationClientStore })
const popOut = async (): Promise<void> => {
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()));
}}
/>
</div>
);
@@ -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<Record<string, ThumbnailState>>;
/** 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<string, SessionObservation>();
private thumbs = new Map<string, ThumbnailState>();
/** sessionId → the attachment id currently displayed for it (leak guard). */
private readonly thumbBySession = new Map<string, string>();
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();
},
@@ -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);
@@ -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);
@@ -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);
}
@@ -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;
}
}
@@ -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 <args...> --json` and collect its output. */
run(args: string[], options?: BskRunOptions): Promise<BskRunResult>;
+77 -42
View File
@@ -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<unknown> {
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");