diff --git a/packages/dsh-plugin-browserskill/src/client/ObservationOverlay.module.css b/packages/dsh-plugin-browserskill/src/client/ObservationOverlay.module.css index 5bc6c41..5093563 100644 --- a/packages/dsh-plugin-browserskill/src/client/ObservationOverlay.module.css +++ b/packages/dsh-plugin-browserskill/src/client/ObservationOverlay.module.css @@ -142,6 +142,8 @@ right: 6px; width: 16px; height: 16px; + padding: 0; + cursor: pointer; border-radius: 50%; display: inline-flex; align-items: center; diff --git a/packages/dsh-plugin-browserskill/src/client/ObservationOverlay.tsx b/packages/dsh-plugin-browserskill/src/client/ObservationOverlay.tsx index a07fe98..1e57704 100644 --- a/packages/dsh-plugin-browserskill/src/client/ObservationOverlay.tsx +++ b/packages/dsh-plugin-browserskill/src/client/ObservationOverlay.tsx @@ -48,7 +48,13 @@ import { createPortal } from "react-dom"; import type { SessionObservation } from "../observation"; import css from "./ObservationOverlay.module.css"; import type { ObservationClientStore } from "./observation-store"; -import { focusOf, statusOf, useObservationView, usePip } from "./observation-view"; +import { + focusOf, + statusOf, + useObservationView, + usePip, + useThumbnailObservation, +} from "./observation-view"; import { getSidebarMode, subscribeSidebarMode } from "./sidebar-mode"; // The pure view helpers live in observation-view (shared with the sidebar @@ -341,6 +347,7 @@ export function OverlayBody(props: { onHeaderPointerDown, } = props; const [interrupting, setInterrupting] = useState(false); + const viewRef = useThumbnailObservation(store, sessions.length > 0); const thumbId = focus?.thumbnailAttachmentId; useEffect(() => { @@ -373,7 +380,12 @@ export function OverlayBody(props: { const state = !available ? "error" : focus !== undefined ? statusOf(focus) : "idle"; return ( -
+
)} {thumb?.status === "error" ? ( - + ) : null}
{sessions.length >= 2 ? ( diff --git a/packages/dsh-plugin-browserskill/src/client/observation-store.ts b/packages/dsh-plugin-browserskill/src/client/observation-store.ts index 2ec6846..4c09ca0 100644 --- a/packages/dsh-plugin-browserskill/src/client/observation-store.ts +++ b/packages/dsh-plugin-browserskill/src/client/observation-store.ts @@ -1,10 +1,9 @@ /** - * Client-side data layer for the observation overlay: initial state fetch, - * 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, + * Client-side data layer for the observation overlay: an ordered SSE snapshot + * and increment stream (each reconnect starts with a fresh snapshot), on-demand + * thumbnail blob loading through the host's ephemeral-frame route, * and the interrupt call. The last ready blob per session is held until its - * replacement decodes so the overlay can swap frames in place. All I/O is + * replacement loads so the overlay can swap frames in place. All I/O is * injected so tests never touch a network. */ @@ -12,8 +11,6 @@ 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; } @@ -50,10 +47,10 @@ export interface OverlaySnapshot { readonly available: boolean; } -const STATE_URL = "/bsk-observation/state"; const EVENTS_URL = "/bsk-observation/events"; const INTERRUPT_URL = "/bsk-observation/interrupt"; const STOP_URL = "/bsk-observation/stop"; +const THUMBNAIL_RETRY_DELAYS_MS = [1000, 3000]; function revoke(url: string | undefined): void { if (url !== undefined && typeof URL.revokeObjectURL === "function") { @@ -66,8 +63,10 @@ export class ObservationClientStore { private thumbs = new Map(); /** sessionId → the attachment id currently advertised for it. */ private readonly thumbBySession = new Map(); - /** sessionId → last successfully decoded frame (held across the next load). */ + /** sessionId → last successfully loaded frame (held across the next load). */ private readonly lastReady = new Map(); + private readonly retryTimers = new Map>(); + private readonly retryCounts = new Map(); private listeners = new Set<() => void>(); private events: EventSourceLike | undefined; private snapshot: OverlaySnapshot = { @@ -81,6 +80,7 @@ export class ObservationClientStore { private started = false; /** Refcount of mounted consumers (overlay card, sidebar tab, sidebar fiber). */ private consumers = 0; + private thumbnailConsumers = 0; constructor(private readonly deps: ObservationClientDeps) {} @@ -133,37 +133,16 @@ export class ObservationClientStore { return undefined; } - /** (Re)pull the full state: initial load and every SSE (re)open. */ - private refreshState(): void { - void this.deps - .fetchFn(STATE_URL) - .then(async (res) => { - if (!res.ok) return; - const body = (await res.json()) as { - sessions?: SessionObservation[]; - available?: boolean; - }; - 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); + private connectEvents(): void { + const previous = this.events; + this.events = undefined; + previous?.close(); + const events = this.deps.eventSourceFactory( + `${EVENTS_URL}?thumbnails=${this.thumbnailConsumers > 0 ? "1" : "0"}`, + ); + this.events = events; events.onmessage = (message) => { + if (!this.started || this.events !== events) return; let event: ObservationEvent; try { event = JSON.parse(message.data) as ObservationEvent; @@ -172,23 +151,24 @@ 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; + } + + /** Every initial connection and reconnect starts with the server's snapshot. */ + start(): void { + if (this.started) return; + this.started = true; + this.connectEvents(); this.publish(); } stop(): void { - this.events?.close(); + const previous = this.events; this.events = undefined; this.started = false; - for (const thumb of this.thumbs.values()) revoke(thumb.url); - this.thumbs.clear(); - this.thumbBySession.clear(); - this.lastReady.clear(); + previous?.close(); + this.clearThumbnails(); this.sessions.clear(); + this.available = true; this.publish(); } @@ -209,6 +189,33 @@ export class ObservationClientStore { if (this.consumers === 0) this.stop(); } + /** Only visible image surfaces request screenshots; metadata watchers do not. */ + watchThumbnails(): () => void { + this.thumbnailConsumers += 1; + if (this.thumbnailConsumers === 1 && this.started) this.connectEvents(); + let released = false; + return () => { + if (released) return; + released = true; + this.thumbnailConsumers -= 1; + if (this.thumbnailConsumers === 0 && this.started) this.connectEvents(); + }; + } + + private forgetThumbnail(attachmentId: string): void { + clearTimeout(this.retryTimers.get(attachmentId)); + this.retryTimers.delete(attachmentId); + this.retryCounts.delete(attachmentId); + revoke(this.thumbs.get(attachmentId)?.url); + this.thumbs.delete(attachmentId); + } + + private clearThumbnails(): void { + for (const attachmentId of this.thumbs.keys()) this.forgetThumbnail(attachmentId); + this.thumbBySession.clear(); + this.lastReady.clear(); + } + /** Forget one session's tracked + held frames, revoking their blob URLs. */ private dropThumb(sessionId: string): void { const attachmentId = this.thumbBySession.get(sessionId); @@ -216,39 +223,45 @@ export class ObservationClientStore { this.thumbBySession.delete(sessionId); this.lastReady.delete(sessionId); if (attachmentId !== undefined) { - revoke(this.thumbs.get(attachmentId)?.url); - this.thumbs.delete(attachmentId); + this.forgetThumbnail(attachmentId); } if (held !== undefined && held.attachmentId !== attachmentId) { - revoke(held.url); - this.thumbs.delete(held.attachmentId); + this.forgetThumbnail(held.attachmentId); } } /** * Track the frame a session currently advertises. The last *ready* blob is - * kept until the replacement decodes — dropping it on the upsert is what + * kept until the replacement loads — dropping it on the upsert is what * made the overlay flash a placeholder between breaths. */ private trackThumb(sessionId: string, attachmentId: string | undefined): void { + if (attachmentId === undefined) { + this.dropThumb(sessionId); + return; + } const previous = this.thumbBySession.get(sessionId); if (previous === attachmentId) return; - if (attachmentId !== undefined) this.thumbBySession.set(sessionId, attachmentId); - else this.thumbBySession.delete(sessionId); + this.thumbBySession.set(sessionId, attachmentId); const heldId = this.lastReady.get(sessionId)?.attachmentId; if (previous !== undefined && previous !== heldId) { - revoke(this.thumbs.get(previous)?.url); - this.thumbs.delete(previous); + this.forgetThumbnail(previous); } } private apply(event: ObservationEvent): void { - if (event.type === "reset") { + if (event.type === "snapshot") { + this.sessions = new Map(event.sessions.map((session) => [session.sessionId, session])); + for (const sessionId of this.thumbBySession.keys()) { + if (!this.sessions.has(sessionId)) this.dropThumb(sessionId); + } + for (const session of event.sessions) { + this.trackThumb(session.sessionId, session.thumbnailAttachmentId); + } + this.available = event.available; + } else if (event.type === "reset") { this.sessions.clear(); - for (const thumb of this.thumbs.values()) revoke(thumb.url); - this.thumbs.clear(); - this.thumbBySession.clear(); - this.lastReady.clear(); + this.clearThumbnails(); } else if (event.type === "remove" && event.session !== undefined) { this.sessions.delete(event.session.sessionId); this.dropThumb(event.session.sessionId); @@ -259,44 +272,77 @@ export class ObservationClientStore { this.available = event.available; } this.publish(); + if (event.type === "snapshot") { + for (const session of event.sessions) this.retryThumbnail(session.thumbnailAttachmentId); + } } /** * Ensure a thumbnail load is in flight for one attachment reference. New - * frames replace the old URL only after they decode; failures keep the last + * frames replace the old URL only after they load; failures keep the last * good frame (the caller renders `status: 'error'` as a small badge over * the old image). */ ensureThumbnail(attachmentId: string | undefined): void { - if (attachmentId === undefined || this.thumbs.has(attachmentId)) return; - this.thumbs.set(attachmentId, { status: "loading" }); + if ( + !this.started || + attachmentId === undefined || + this.sessionOf(attachmentId) === undefined || + this.thumbs.has(attachmentId) + ) + return; + const loading: ThumbnailState = { status: "loading" }; + this.thumbs.set(attachmentId, loading); this.publish(); this.deps.loadImage(attachmentId).then( (url) => { - if (!this.thumbs.has(attachmentId)) { + if (this.thumbs.get(attachmentId) !== loading) { // Replaced or removed while loading: never resurrect (or leak) it. revoke(url); return; } + this.retryCounts.delete(attachmentId); this.thumbs.set(attachmentId, { status: "ready", url }); const sessionId = this.sessionOf(attachmentId); if (sessionId !== undefined) { const previous = this.lastReady.get(sessionId); if (previous !== undefined && previous.attachmentId !== attachmentId) { - revoke(previous.url); - this.thumbs.delete(previous.attachmentId); + this.forgetThumbnail(previous.attachmentId); } this.lastReady.set(sessionId, { attachmentId, url }); } this.publish(); }, () => { - this.thumbs.set(attachmentId, { status: "error" }); + if (this.thumbs.get(attachmentId) !== loading) return; + const failed: ThumbnailState = { status: "error" }; + this.thumbs.set(attachmentId, failed); + const attempts = this.retryCounts.get(attachmentId) ?? 0; + const delay = THUMBNAIL_RETRY_DELAYS_MS[attempts]; + if (delay !== undefined) { + this.retryCounts.set(attachmentId, attempts + 1); + this.retryTimers.set( + attachmentId, + setTimeout(() => { + this.retryTimers.delete(attachmentId); + if (this.thumbs.get(attachmentId) !== failed) return; + this.thumbs.delete(attachmentId); + this.ensureThumbnail(attachmentId); + }, delay), + ); + } this.publish(); }, ); } + /** Explicit retry (also used on a fresh connection after an outage). */ + retryThumbnail(attachmentId: string | undefined): void { + if (attachmentId === undefined || this.thumbs.get(attachmentId)?.status !== "error") return; + this.forgetThumbnail(attachmentId); + this.ensureThumbnail(attachmentId); + } + /** Interrupt the current or named session's in-flight call. */ async interrupt(sessionId?: string): Promise { try { diff --git a/packages/dsh-plugin-browserskill/src/client/observation-view.ts b/packages/dsh-plugin-browserskill/src/client/observation-view.ts index 987ddc6..e10173c 100644 --- a/packages/dsh-plugin-browserskill/src/client/observation-view.ts +++ b/packages/dsh-plugin-browserskill/src/client/observation-view.ts @@ -6,7 +6,7 @@ * without duplicating hooks. */ -import { useCallback, useEffect, useState, useSyncExternalStore } from "react"; +import { useCallback, useEffect, useRef, useState, useSyncExternalStore } from "react"; import type { SessionObservation } from "../observation"; import type { ObservationClientStore, OverlaySnapshot } from "./observation-store"; @@ -114,6 +114,49 @@ export function useObservationView( }; } +/** A mounted metadata watcher is not necessarily a visible screenshot viewer. */ +export function useThumbnailObservation(store: ObservationClientStore, enabled: boolean) { + const ref = useRef(null); + useEffect(() => { + const element = ref.current; + if (!enabled || element === null) return; + // A portal uses its PiP document's visibility, independently of the main tab. + const doc = element.ownerDocument; + const Observer = doc.defaultView?.IntersectionObserver; + let active = true; + let intersecting = Observer === undefined; + let release: (() => void) | undefined; + const update = () => { + if (!active) return; + const visible = intersecting && doc.visibilityState !== "hidden"; + if (visible && release === undefined) release = store.watchThumbnails(); + else if (!visible && release !== undefined) { + release(); + release = undefined; + } + }; + const observer = + Observer === undefined + ? undefined + : new Observer((entries) => { + intersecting = entries.some( + (entry) => entry.target === element && entry.isIntersecting, + ); + update(); + }); + observer?.observe(element); + doc.addEventListener("visibilitychange", update); + update(); + return () => { + active = false; + observer?.disconnect(); + doc.removeEventListener("visibilitychange", update); + release?.(); + }; + }, [store, enabled]); + return ref; +} + export interface PipHandle { /** The open PiP window, or null while the content renders in its carrier. */ readonly pipWindow: Window | null; diff --git a/packages/dsh-plugin-browserskill/src/observation-http.ts b/packages/dsh-plugin-browserskill/src/observation-http.ts index 09add13..115d689 100644 --- a/packages/dsh-plugin-browserskill/src/observation-http.ts +++ b/packages/dsh-plugin-browserskill/src/observation-http.ts @@ -6,7 +6,8 @@ * observation channel through the documented `webServer` route seam instead: * * GET /bsk-observation/state → { sessions, available } - * GET /bsk-observation/events → SSE stream of ObservationEvent + * GET /bsk-observation/events → SSE snapshot, then ObservationEvent increments + * (?thumbnails=0 for state-only subscribers) * POST /bsk-observation/interrupt → body {sessionId?} → {interrupted: boolean} * POST /bsk-observation/stop → body {sessionId} → {stopped: boolean} * GET /bsk-observation/thumbnail/ → image bytes @@ -99,6 +100,7 @@ export function registerObservationRoutes( if (webServer === undefined) { return () => {}; } + const streams = new Set<() => void>(); const disposers = [ webServer.register({ kind: "exact", @@ -129,14 +131,42 @@ export function registerObservationRoutes( "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); - res.on("close", () => { + const thumbnails = + new URL(req.url ?? "/", "http://localhost").searchParams.get("thumbnails") !== "0"; + let unsubscribe: (() => void) | undefined; + let heartbeat: ReturnType | undefined; + let closed = false; + const close = () => { + if (closed) return; + closed = true; clearInterval(heartbeat); - unsubscribe(); - }); + unsubscribe?.(); + streams.delete(close); + res.end(); + }; + const write = (data: string) => { + if (closed) return; + try { + res.write(data); + } catch { + close(); + } + }; + streams.add(close); + res.on("close", close); + res.on("error", close); + try { + unsubscribe = observation.subscribe( + (event) => { + write(`data: ${JSON.stringify(event)}\n\n`); + }, + { thumbnails }, + ); + if (closed) unsubscribe(); + else heartbeat = setInterval(() => write(": heartbeat\n\n"), SSE_HEARTBEAT_MS); + } catch { + close(); + } }, }), webServer.register({ @@ -228,6 +258,7 @@ export function registerObservationRoutes( }), ]; return () => { + for (const close of streams) close(); for (const dispose of disposers) dispose(); }; } diff --git a/packages/dsh-plugin-browserskill/src/observation.ts b/packages/dsh-plugin-browserskill/src/observation.ts index 5c280f2..17b0194 100644 --- a/packages/dsh-plugin-browserskill/src/observation.ts +++ b/packages/dsh-plugin-browserskill/src/observation.ts @@ -49,8 +49,9 @@ export interface SessionObservation { dshSessionIds?: string[]; } -/** Incremental event carried to subscribers (SSE on the wire). */ +/** Initial snapshot or incremental change carried to subscribers (SSE on the wire). */ export type ObservationEvent = + | { type: "snapshot"; sessions: SessionObservation[]; available: boolean } | { type: "upsert" | "remove" | "reset"; session?: SessionObservation } | { type: "availability"; available: boolean }; @@ -89,6 +90,7 @@ export class ObservationService { private readonly scratchNamespace = randomUUID(); private readonly observations = new Map(); private readonly listeners = new Set<(event: ObservationEvent) => void>(); + private thumbnailViewers = 0; private readonly captureTimers = new Map(); private readonly captureInFlight = new Set(); /** Captures intentionally cancelled to make way for foreground work. */ @@ -141,10 +143,30 @@ export class ObservationService { this.emit({ type: "availability", available }); } - /** Subscribe to incremental changes; returns an unsubscribe function. */ - subscribe(listener: (event: ObservationEvent) => void): () => void { - this.listeners.add(listener); - return () => this.listeners.delete(listener); + /** A snapshot and subsequent changes share one ordered subscription. */ + subscribe( + listener: (event: ObservationEvent) => void, + { thumbnails = true }: { thumbnails?: boolean } = {}, + ): () => void { + if (this.disposed) return () => {}; + // Each subscription owns its lease, even if a caller reuses a callback. + const receive = (event: ObservationEvent) => listener(event); + this.listeners.add(receive); + try { + listener({ type: "snapshot", sessions: this.getState(), available: this.available }); + } catch (error) { + this.listeners.delete(receive); + throw error; + } + if (thumbnails && ++this.thumbnailViewers === 1) { + for (const sessionId of this.observations.keys()) this.scheduleCapture(sessionId, 0); + } + return () => { + if (!this.listeners.delete(receive)) return; + if (thumbnails && --this.thumbnailViewers === 0) { + for (const sessionId of this.captureTimers.keys()) this.cancelCapture(sessionId); + } + }; } private emit(event: ObservationEvent): void { @@ -355,6 +377,7 @@ export class ObservationService { this.observations.clear(); this.emit({ type: "reset" }); this.listeners.clear(); + this.thumbnailViewers = 0; } // ------------------------------------------------------------------ @@ -369,12 +392,21 @@ export class ObservationService { } } + private canCapture(sessionId: string): boolean { + const entry = this.observations.get(sessionId); + return ( + this.deps.options.enabled && + !this.disposed && + this.thumbnailViewers > 0 && + !this.foregroundDepth.has(sessionId) && + entry !== undefined && + entry.dead !== true + ); + } + /** 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.foregroundDepth.has(sessionId)) return; - const entry = this.observations.get(sessionId); - if (entry === undefined || entry.dead === true) return; + if (!this.canCapture(sessionId)) return; this.cancelCapture(sessionId); const now = this.scheduler.now(); const lastSeen = this.lastActivity.get(sessionId) ?? 0; @@ -399,9 +431,7 @@ export class ObservationService { * Failures keep the previous frame and back off silently. */ private async capture(sessionId: string): Promise { - if (this.disposed) return; - const current = this.observations.get(sessionId); - if (current === undefined || current.dead === true) return; + if (!this.canCapture(sessionId)) return; if (this.captureInFlight.has(sessionId)) return; this.captureInFlight.add(sessionId); // Stable within this service, isolated from captures owned by other instances. @@ -410,8 +440,8 @@ export class ObservationService { let writtenPath = outPath; try { const result = await this.deps.queue.run(sessionId, () => { - // A foreground lease may have arrived while this capture was queued. - if (this.foregroundDepth.has(sessionId)) return Promise.resolve(undefined); + // The last viewer may have left, or foreground work arrived, while queued. + if (!this.canCapture(sessionId)) return Promise.resolve(undefined); return this.deps.runner.run(["screenshot", "--session", sessionId, "--out", outPath], { timeoutMs: 15_000, tag: `observation:${sessionId}`, @@ -445,6 +475,7 @@ export class ObservationService { this.setAvailable(true); this.publishFrame(sessionId, entry, data); } catch { + if (this.disposed || !this.observations.has(sessionId)) return; // Foreground preemption is healthy scheduling, not browser/daemon // unavailability; do not let normal tool traffic trip capture backoff. if (!this.capturePreempted.delete(sessionId)) { diff --git a/packages/dsh-plugin-browserskill/tests/client/observation-overlay.test.tsx b/packages/dsh-plugin-browserskill/tests/client/observation-overlay.test.tsx index 21cbdbd..e257397 100644 --- a/packages/dsh-plugin-browserskill/tests/client/observation-overlay.test.tsx +++ b/packages/dsh-plugin-browserskill/tests/client/observation-overlay.test.tsx @@ -19,10 +19,12 @@ interface Harness { emitRaw: (event: unknown) => void; fetches: { url: string; init?: { body?: string } }[]; loadImage: ReturnType; + eventUrls: string[]; } function makeHarness(initial: SessionObservation[]): Harness { const fetches: Harness["fetches"] = []; + const eventUrls: string[] = []; let es: EventSourceLike | undefined; let current = initial; const loadImage = vi.fn(async (id: string) => `blob:${id}`); @@ -34,9 +36,16 @@ function makeHarness(initial: SessionObservation[]): Harness { } return { ok: true, json: async () => ({ interrupted: true }) }; }, - eventSourceFactory: () => { - es = { onmessage: null, close: vi.fn() }; - return es; + eventSourceFactory: (url) => { + eventUrls.push(url); + const connection = { onmessage: null, close: vi.fn() } as EventSourceLike; + es = connection; + queueMicrotask(() => + connection.onmessage?.({ + data: JSON.stringify({ type: "snapshot", sessions: current, available: true }), + }), + ); + return connection; }, loadImage, }); @@ -58,10 +67,120 @@ function makeHarness(initial: SessionObservation[]): Harness { }, fetches, loadImage, + eventUrls, }; } -afterEach(cleanup); +afterEach(() => { + cleanup(); + vi.restoreAllMocks(); + vi.unstubAllGlobals(); +}); + +describe("visible thumbnail demand", () => { + function intersections() { + const observers: Array<{ + notify: (visible: boolean) => void; + disconnect: ReturnType; + }> = []; + vi.stubGlobal( + "IntersectionObserver", + class { + private target?: Element; + disconnect = vi.fn(); + constructor(private callback: IntersectionObserverCallback) { + observers.push(this); + } + observe(target: Element) { + this.target = target; + } + notify(visible: boolean) { + this.callback( + [{ target: this.target, isIntersecting: visible } as IntersectionObserverEntry], + this as unknown as IntersectionObserver, + ); + } + }, + ); + return observers; + } + + it("pauses screenshots on collapse while keeping the metadata feed", async () => { + const observers = intersections(); + const h = makeHarness([BUSY]); + const { unmount } = render(); + await screen.findByTestId("obs-card"); + expect(h.eventUrls.at(-1)).toBe("/bsk-observation/events?thumbnails=0"); + act(() => observers[0].notify(true)); + expect(h.eventUrls.at(-1)).toBe("/bsk-observation/events?thumbnails=1"); + fireEvent.click(screen.getByRole("button", { name: "Collapse" })); + expect(h.eventUrls.at(-1)).toBe("/bsk-observation/events?thumbnails=0"); + expect(h.es().close).not.toHaveBeenCalled(); + expect(observers[0].disconnect).toHaveBeenCalledOnce(); + // A callback queued before disconnect must not reclaim screenshot demand. + act(() => observers[0].notify(true)); + expect(h.eventUrls.at(-1)).toBe("/bsk-observation/events?thumbnails=0"); + fireEvent.click(screen.getByRole("button", { name: /Expand browser observation/ })); + act(() => observers[1].notify(true)); + expect(h.eventUrls.at(-1)).toBe("/bsk-observation/events?thumbnails=1"); + unmount(); + expect(h.es().close).toHaveBeenCalledOnce(); + }); + + it("ignores an initial visibility notification delivered after collapse", async () => { + const observers = intersections(); + const h = makeHarness([BUSY]); + render(); + await screen.findByTestId("obs-card"); + fireEvent.click(screen.getByRole("button", { name: "Collapse" })); + act(() => observers[0].notify(true)); + expect(h.eventUrls).toEqual(["/bsk-observation/events?thumbnails=0"]); + }); + + it("pauses for a hidden document or hidden panel and resumes when visible", async () => { + const observers = intersections(); + let visibility: DocumentVisibilityState = "visible"; + vi.spyOn(document, "visibilityState", "get").mockImplementation(() => visibility); + const h = makeHarness([BUSY]); + render(); + await screen.findByTestId("obs-card"); + act(() => observers[0].notify(true)); + expect(h.eventUrls.at(-1)).toContain("thumbnails=1"); + act(() => { + visibility = "hidden"; + document.dispatchEvent(new Event("visibilitychange")); + }); + expect(h.eventUrls.at(-1)).toContain("thumbnails=0"); + act(() => { + visibility = "visible"; + document.dispatchEvent(new Event("visibilitychange")); + }); + expect(h.eventUrls.at(-1)).toContain("thumbnails=1"); + act(() => observers[0].notify(false)); + expect(h.eventUrls.at(-1)).toContain("thumbnails=0"); + }); + + it("keeps a visible PiP watching when the main document is hidden", async () => { + // The detached PiP test document has no window/IntersectionObserver. + vi.stubGlobal("IntersectionObserver", undefined); + let visibility: DocumentVisibilityState = "visible"; + vi.spyOn(document, "visibilityState", "get").mockImplementation(() => visibility); + const pipDoc = document.implementation.createHTMLDocument("pip"); + Object.defineProperty(pipDoc, "visibilityState", { value: "visible" }); + (window as unknown as Record).documentPictureInPicture = { + requestWindow: async () => ({ document: pipDoc, addEventListener: () => {}, close: vi.fn() }), + }; + const h = makeHarness([BUSY]); + render(); + fireEvent.click(await screen.findByRole("button", { name: /Pop out/ })); + await waitFor(() => expect(pipDoc.body.textContent).toContain("s1")); + act(() => { + visibility = "hidden"; + document.dispatchEvent(new Event("visibilitychange")); + }); + expect(h.eventUrls.at(-1)).toContain("thumbnails=1"); + }); +}); describe("ObservationOverlay", () => { beforeEach(() => { @@ -92,6 +211,16 @@ describe("ObservationOverlay", () => { await screen.findByRole("img", { name: "session s1 view" }); }); + it("offers a retry after a thumbnail fails without requiring a new frame id", async () => { + const h = makeHarness([{ ...BUSY, thumbnailAttachmentId: "att-9" }]); + h.loadImage.mockRejectedValueOnce(new Error("temporary failure")); + render(); + fireEvent.click(await screen.findByRole("button", { name: "Retry thumbnail" })); + await screen.findByRole("img", { name: "session s1 view" }); + expect(h.loadImage).toHaveBeenCalledTimes(2); + expect(screen.queryByRole("button", { name: "Retry thumbnail" })).toBeNull(); + }); + it("keeps the current frame on stage while the next thumbnail loads", async () => { let resolveNext: ((url: string) => void) | undefined; const h = makeHarness([{ ...BUSY, thumbnailAttachmentId: "att-1" }]); diff --git a/packages/dsh-plugin-browserskill/tests/client/observation-sidebar.test.tsx b/packages/dsh-plugin-browserskill/tests/client/observation-sidebar.test.tsx index efbfb0a..cd95f62 100644 --- a/packages/dsh-plugin-browserskill/tests/client/observation-sidebar.test.tsx +++ b/packages/dsh-plugin-browserskill/tests/client/observation-sidebar.test.tsx @@ -53,8 +53,14 @@ function makeHarness(initial: SessionObservation[]): Harness { return { ok: true, json: async () => ({ interrupted: true }) }; }, eventSourceFactory: () => { - es = { onmessage: null, close: vi.fn() }; - return es; + const connection = { onmessage: null, close: vi.fn() } as EventSourceLike; + es = connection; + queueMicrotask(() => + connection.onmessage?.({ + data: JSON.stringify({ type: "snapshot", sessions: current, available: true }), + }), + ); + return connection; }, loadImage: async (id: string) => `blob:${id}`, }); 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 37e4d47..aab7229 100644 --- a/packages/dsh-plugin-browserskill/tests/client/observation-store.test.ts +++ b/packages/dsh-plugin-browserskill/tests/client/observation-store.test.ts @@ -1,12 +1,19 @@ -// ObservationClientStore: initial fetch + SSE application, thumbnail loading +// ObservationClientStore: ordered SSE snapshots/increments, thumbnail loading // lifecycle, interrupt wire shape. All I/O faked. -import { describe, expect, it, vi } from "vitest"; +import { afterEach, describe, expect, it, vi } from "vitest"; import { type EventSourceLike, ObservationClientStore } from "../../src/client/observation-store"; import type { ObservationEvent, SessionObservation } from "../../src/observation"; const OBS_IDLE: SessionObservation = { sessionId: "s1", action: "idle", since: 1000 }; const OBS_BUSY: SessionObservation = { sessionId: "s1", action: "clicking", since: 2000 }; +const stores: ObservationClientStore[] = []; + +afterEach(() => { + for (const store of stores.splice(0)) store.stop(); + vi.restoreAllMocks(); + vi.useRealTimers(); +}); function harness(state: SessionObservation[] = []) { const fetches: { @@ -14,6 +21,7 @@ function harness(state: SessionObservation[] = []) { init?: { method?: string; body?: string; headers?: Record }; }[] = []; const esInstances: EventSourceLike[] = []; + const eventUrls: string[] = []; const fetchFn = vi.fn( async ( url: string, @@ -29,11 +37,14 @@ function harness(state: SessionObservation[] = []) { const eventSourceFactory = (url: string) => { const es: EventSourceLike = { onmessage: null, close: vi.fn() }; esInstances.push(es); + eventUrls.push(url); + queueMicrotask(() => emit(es, { type: "snapshot", sessions: state, available: true })); return es; }; const loadImage = vi.fn(async (id: string) => `blob:url-${id}`); const store = new ObservationClientStore({ fetchFn, eventSourceFactory, loadImage }); - return { store, fetches, esInstances, loadImage }; + stores.push(store); + return { store, fetches, esInstances, eventUrls, loadImage }; } function emit(es: EventSourceLike, event: ObservationEvent): void { @@ -73,7 +84,9 @@ describe("ObservationClientStore", () => { }); it("loads each thumbnail once and reports ready", async () => { - const { store, loadImage } = harness(); + const { store, loadImage } = harness([{ ...OBS_IDLE, thumbnailAttachmentId: "att-1" }]); + store.start(); + await Promise.resolve(); store.ensureThumbnail("att-1"); store.ensureThumbnail("att-1"); expect(loadImage).toHaveBeenCalledTimes(1); @@ -86,7 +99,9 @@ describe("ObservationClientStore", () => { }); it("marks failed thumbnail loads as error", async () => { - const { store, loadImage } = harness(); + const { store, loadImage } = harness([{ ...OBS_IDLE, thumbnailAttachmentId: "att-bad" }]); + store.start(); + await Promise.resolve(); loadImage.mockRejectedValueOnce(new Error("nope")); store.ensureThumbnail("att-bad"); await vi.waitFor(() => expect(store.getSnapshot().thumbnails["att-bad"].status).toBe("error")); @@ -124,9 +139,7 @@ describe("ObservationClientStore", () => { describe("thumbnail blob URL lifecycle", () => { function withRevokeSpy() { - const revoke = vi.fn(); - (URL as unknown as { revokeObjectURL: unknown }).revokeObjectURL = revoke; - return revoke; + return vi.spyOn(URL, "revokeObjectURL").mockImplementation(() => {}); } it("holds the last ready frame until the replacement decodes", async () => { @@ -231,16 +244,216 @@ describe("thumbnail blob URL lifecycle", () => { }); }); -describe("SSE (re)open resync", () => { - it("refetches the full state when the stream (re)opens", async () => { +describe("SSE snapshot ordering", () => { + it("ignores events from a stopped connection after a restart", async () => { + const { store, esInstances } = harness([OBS_IDLE]); + store.start(); + await Promise.resolve(); + const old = esInstances[0]; + store.stop(); + emit(old, { type: "upsert", session: OBS_BUSY }); + expect(store.getSnapshot().sessions).toEqual([]); + store.start(); + await Promise.resolve(); + emit(esInstances[1], { type: "upsert", session: OBS_BUSY }); + emit(old, { type: "snapshot", sessions: [], available: false }); + emit(old, { type: "remove", session: OBS_BUSY }); + expect(store.getSnapshot()).toMatchObject({ sessions: [OBS_BUSY], available: true }); + }); + + it("does not apply a queued initial snapshot after stop", async () => { + const { store } = harness([OBS_BUSY]); + store.start(); + store.stop(); + await Promise.resolve(); + expect(store.getSnapshot()).toMatchObject({ sessions: [], thumbnails: {}, subscribed: false }); + }); + + it("replaces state on reconnect before applying subsequent events, without a competing fetch", 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), - ); + emit(esInstances[0], { type: "snapshot", sessions: [], available: false }); + expect(store.getSnapshot()).toMatchObject({ sessions: [], available: false }); + emit(esInstances[0], { type: "upsert", session: OBS_BUSY }); + await Promise.resolve(); + expect(store.getSnapshot().sessions).toEqual([OBS_BUSY]); + expect(fetches).toEqual([]); store.stop(); }); }); + +describe("thumbnail recovery and stale loads", () => { + function deferred() { + let resolve!: (value: T) => void; + let reject!: (reason: Error) => void; + const promise = new Promise((yes, no) => { + resolve = yes; + reject = no; + }); + return { promise, resolve, reject }; + } + + it("retries a transient failure without needing a different frame id", async () => { + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] }); + const { store, loadImage } = harness([{ ...OBS_IDLE, thumbnailAttachmentId: "a1" }]); + store.start(); + await Promise.resolve(); + loadImage.mockRejectedValueOnce(new Error("temporary")); + store.ensureThumbnail("a1"); + await Promise.resolve(); + expect(store.getSnapshot().thumbnails.a1.status).toBe("error"); + await vi.advanceTimersByTimeAsync(1000); + expect(loadImage).toHaveBeenCalledTimes(2); + expect(store.getSnapshot().displayFrames.s1).toEqual({ status: "ready", url: "blob:url-a1" }); + expect(vi.getTimerCount()).toBe(0); + }); + + it("bounds automatic retries and allows an explicit recovery", async () => { + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] }); + const { store, loadImage } = harness([{ ...OBS_IDLE, thumbnailAttachmentId: "a1" }]); + store.start(); + await Promise.resolve(); + loadImage.mockRejectedValue(new Error("offline")); + store.ensureThumbnail("a1"); + await vi.advanceTimersByTimeAsync(60_000); + expect(loadImage).toHaveBeenCalledTimes(3); + expect(vi.getTimerCount()).toBe(0); + store.ensureThumbnail("a1"); + expect(loadImage).toHaveBeenCalledTimes(3); + loadImage.mockResolvedValue("blob:recovered"); + store.retryThumbnail("a1"); + await Promise.resolve(); + expect(store.getSnapshot().displayFrames.s1.url).toBe("blob:recovered"); + }); + + it("cancels backoff when a reconnect retries the current frame", async () => { + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] }); + const session = { ...OBS_IDLE, thumbnailAttachmentId: "a1" }; + const { store, esInstances, loadImage } = harness([session]); + store.start(); + await Promise.resolve(); + loadImage.mockRejectedValueOnce(new Error("offline")); + store.ensureThumbnail("a1"); + await Promise.resolve(); + emit(esInstances[0], { type: "snapshot", sessions: [session], available: true }); + await Promise.resolve(); + expect(store.getSnapshot().thumbnails.a1.status).toBe("ready"); + await vi.advanceTimersByTimeAsync(60_000); + expect(loadImage).toHaveBeenCalledTimes(2); + }); + + it.each([ + "remove", + "reset", + "stop", + "missing-thumbnail", + "snapshot-removal", + ])("cleans frames and cancels retries on %s", async (action) => { + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] }); + const revoke = vi.spyOn(URL, "revokeObjectURL").mockImplementation(() => {}); + const { store, esInstances, loadImage } = harness([ + { ...OBS_IDLE, thumbnailAttachmentId: "a1" }, + ]); + store.start(); + await Promise.resolve(); + store.ensureThumbnail("a1"); + await Promise.resolve(); + emit(esInstances[0], { type: "upsert", session: { ...OBS_IDLE, thumbnailAttachmentId: "a2" } }); + loadImage.mockRejectedValueOnce(new Error("no frame")); + store.ensureThumbnail("a2"); + await Promise.resolve(); + expect(store.getSnapshot().displayFrames.s1).toEqual({ status: "error", url: "blob:url-a1" }); + if (action === "stop") store.stop(); + else if (action === "remove") emit(esInstances[0], { type: "remove", session: OBS_IDLE }); + else if (action === "reset") emit(esInstances[0], { type: "reset" }); + else if (action === "missing-thumbnail") + emit(esInstances[0], { type: "upsert", session: OBS_IDLE }); + else emit(esInstances[0], { type: "snapshot", sessions: [], available: true }); + expect(store.getSnapshot().thumbnails).toEqual({}); + expect(store.getSnapshot().displayFrames).toEqual({}); + expect(revoke).toHaveBeenCalledExactlyOnceWith("blob:url-a1"); + expect(vi.getTimerCount()).toBe(0); + await vi.advanceTimersByTimeAsync(60_000); + expect(loadImage).toHaveBeenCalledTimes(2); + }); + + it.each([ + "resolve", + "reject", + ])("ignores a stale load's %s even when the same id is loaded again", async (outcome) => { + const revoke = vi.spyOn(URL, "revokeObjectURL").mockImplementation(() => {}); + const { store, esInstances, loadImage } = harness([ + { ...OBS_IDLE, thumbnailAttachmentId: "a1" }, + ]); + const old = deferred(); + loadImage.mockReturnValueOnce(old.promise); + store.start(); + await Promise.resolve(); + store.ensureThumbnail("a1"); + emit(esInstances[0], { type: "upsert", session: { ...OBS_IDLE, thumbnailAttachmentId: "a2" } }); + emit(esInstances[0], { type: "upsert", session: { ...OBS_IDLE, thumbnailAttachmentId: "a1" } }); + store.ensureThumbnail("a1"); + await Promise.resolve(); + if (outcome === "resolve") old.resolve("blob:stale"); + else old.reject(new Error("stale error")); + await Promise.resolve(); + expect(store.getSnapshot().displayFrames.s1).toEqual({ status: "ready", url: "blob:url-a1" }); + expect(revoke).not.toHaveBeenCalledWith("blob:url-a1"); + if (outcome === "resolve") expect(revoke).toHaveBeenCalledWith("blob:stale"); + }); + + it.each(["resolve", "reject"])("ignores an in-flight load's %s after stop", async (outcome) => { + const revoke = vi.spyOn(URL, "revokeObjectURL").mockImplementation(() => {}); + const { store, loadImage } = harness([{ ...OBS_IDLE, thumbnailAttachmentId: "a1" }]); + const pending = deferred(); + loadImage.mockReturnValueOnce(pending.promise); + store.start(); + await Promise.resolve(); + store.ensureThumbnail("a1"); + store.stop(); + if (outcome === "resolve") pending.resolve("blob:late"); + else pending.reject(new Error("late error")); + await Promise.resolve(); + expect(store.getSnapshot()).toMatchObject({ sessions: [], thumbnails: {}, displayFrames: {} }); + if (outcome === "resolve") expect(revoke).toHaveBeenCalledWith("blob:late"); + }); +}); + +describe("thumbnail viewer leases", () => { + it("keeps metadata subscribed while the last screenshot viewer leaves", async () => { + const { store, eventUrls, esInstances } = harness([OBS_IDLE]); + store.acquire(); + await Promise.resolve(); + expect(eventUrls).toEqual(["/bsk-observation/events?thumbnails=0"]); + const first = store.watchThumbnails(); + expect(eventUrls.at(-1)).toBe("/bsk-observation/events?thumbnails=1"); + const second = store.watchThumbnails(); + expect(eventUrls).toHaveLength(2); + first(); + first(); + expect(eventUrls).toHaveLength(2); + second(); + expect(eventUrls.at(-1)).toBe("/bsk-observation/events?thumbnails=0"); + expect(store.getSnapshot().subscribed).toBe(true); + store.release(); + expect(esInstances.at(-1)?.close).toHaveBeenCalledOnce(); + expect(store.getSnapshot().subscribed).toBe(false); + }); + + it("ignores delayed snapshots from a replaced stream without clearing the displayed state", async () => { + const { store, esInstances } = harness([OBS_IDLE]); + store.start(); + await Promise.resolve(); + emit(esInstances[0], { type: "upsert", session: OBS_BUSY }); + const release = store.watchThumbnails(); + expect(store.getSnapshot().sessions).toEqual([OBS_BUSY]); + await Promise.resolve(); + emit(esInstances[1], { type: "upsert", session: OBS_BUSY }); + emit(esInstances[0], { type: "snapshot", sessions: [], available: false }); + expect(store.getSnapshot()).toMatchObject({ sessions: [OBS_BUSY], available: true }); + store.stop(); + release(); + expect(esInstances).toHaveLength(2); + }); +}); diff --git a/packages/dsh-plugin-browserskill/tests/observation.test.ts b/packages/dsh-plugin-browserskill/tests/observation.test.ts index 05b2eb2..af857b2 100644 --- a/packages/dsh-plugin-browserskill/tests/observation.test.ts +++ b/packages/dsh-plugin-browserskill/tests/observation.test.ts @@ -6,7 +6,7 @@ import { existsSync, mkdtempSync, writeFileSync } from "node:fs"; import type { IncomingMessage, ServerResponse } from "node:http"; import { tmpdir } from "node:os"; import { basename, dirname } from "node:path"; -import { describe, expect, it } from "vitest"; +import { describe, expect, it, vi } from "vitest"; import { actionForLabel, type ObservationEvent, @@ -138,6 +138,8 @@ function setup(opts: { runner?: FakeRunner; registry?: SessionRegistry; scheduler?: ReturnType; + thumbnails?: boolean; + queue?: KeyedExecutor; }) { const registry = opts.registry ?? new SessionRegistry(5); const runner = opts.runner ?? fakeRunner(); @@ -146,13 +148,18 @@ function setup(opts: { ctx: fakeCtx(), runner, registry, - queue: new KeyedExecutor(), + queue: opts.queue ?? new KeyedExecutor(), options: OPTIONS, scheduler, }); const events: ObservationEvent[] = []; - service.subscribe((event) => events.push(event)); - return { service, registry, runner, scheduler, events }; + const unsubscribe = service.subscribe( + (event) => { + if (event.type !== "snapshot") events.push(event); + }, + { thumbnails: opts.thumbnails ?? true }, + ); + return { service, registry, runner, scheduler, events, unsubscribe }; } /** Drive the registry through a real start so the session is owned. */ @@ -224,6 +231,111 @@ describe("state machine", () => { }); describe("thumbnail cadence", () => { + it("keeps metadata current without screenshots and resumes when a viewer arrives", async () => { + const { service, scheduler, runner, events } = setup({ thumbnails: false }); + service.addSession("s1"); + service.beginAction("s1", "clicking"); + service.endAction("s1"); + const releaseForeground = service.acquireForeground("s1"); + releaseForeground(); + expect(service.getState()[0].action).toBe("idle"); + expect(events).toHaveLength(3); + expect(scheduler.pending()).toEqual([]); + expect(runner.calls).toEqual([]); + const snapshots: ObservationEvent[] = []; + const leave = service.subscribe((event) => snapshots.push(event)); + expect(snapshots[0]).toEqual({ + type: "snapshot", + sessions: service.getState(), + available: true, + }); + expect(scheduler.pending()).toEqual([0]); + scheduler.runNext(); + await waitFor(() => scheduler.pending().length === 1); + expect(runner.calls).toHaveLength(1); + leave(); + leave(); + expect(scheduler.pending()).toEqual([]); + service.endAction("s1"); + expect(scheduler.pending()).toEqual([]); + const resume = service.subscribe(() => {}); + expect(scheduler.pending()).toEqual([0]); + resume(); + service.dispose(); + }); + + it("does not capture in a headless service with no subscribers", () => { + const { service, scheduler, unsubscribe } = setup({}); + unsubscribe(); + service.addSession("s1"); + service.endAction("s1"); + expect(scheduler.pending()).toEqual([]); + expect(service.getState()).toHaveLength(1); + service.dispose(); + }); + + it("keeps capturing until the last viewer leaves, including reused callbacks", () => { + const { service, scheduler } = setup({ thumbnails: false }); + service.addSession("s1"); + const listener = () => {}; + const first = service.subscribe(listener); + const second = service.subscribe(listener); + expect(scheduler.pending()).toEqual([0]); + first(); + first(); + expect(scheduler.pending()).toEqual([0]); + second(); + expect(scheduler.pending()).toEqual([]); + service.dispose(); + }); + + it("skips a queued screenshot if its last viewer leaves before execution", async () => { + const queue = new KeyedExecutor(); + let unblock!: () => void; + const gate = new Promise((resolve) => { + unblock = resolve; + }); + const busy = queue.run("s1", () => gate); + const { service, scheduler, runner, unsubscribe } = setup({ queue }); + service.addSession("s1"); + scheduler.runNext(); + unsubscribe(); + unblock(); + await busy; + await queue.run("s1", async () => {}); + expect(runner.calls).toEqual([]); + expect(scheduler.pending()).toEqual([]); + service.dispose(); + }); + + it("lets an already running capture finish without restarting the loop after the last viewer leaves", async () => { + const runner = fakeRunner(); + const run = runner.run.bind(runner); + let finish!: () => void; + let started = false; + runner.run = async (args, options) => { + started = true; + await new Promise((resolve) => { + finish = resolve; + }); + return run(args, options); + }; + const { service, scheduler, unsubscribe } = setup({ runner }); + service.addSession("s1"); + scheduler.runNext(); + await waitFor(() => started); + unsubscribe(); + finish(); + await waitFor(() => service.getState()[0]?.thumbnailAttachmentId !== undefined); + const args = runner.calls[0].args; + await waitFor(() => !existsSync(args[args.indexOf("--out") + 1])); + expect(scheduler.pending()).toEqual([]); + const leave = service.subscribe(() => {}); + expect(scheduler.pending()).toEqual([0]); + leave(); + service.dispose(); + }); + it("gives foreground work priority and resumes with one fresh capture", async () => { const scheduler = fakeScheduler(); let finishCapture: ((result: BskRunResult) => void) | undefined; @@ -540,6 +652,12 @@ describe("HTTP/SSE interface", () => { const eventsRes = fakeRes(); await eventsRoute?.handler(fakeReq(), eventsRes.res as never); service.beginAction("s1", "clicking"); + const streamed = eventsRes.chunks.map((chunk) => JSON.parse(chunk.slice(6))); + expect(streamed.map((event) => event.type)).toEqual(["snapshot", "upsert"]); + expect(streamed[0]).toMatchObject({ + sessions: [{ sessionId: "s1", action: "idle" }], + available: true, + }); expect(eventsRes.res.body()).toContain('"action":"clicking"'); eventsRes.res.close(); @@ -566,6 +684,47 @@ describe("HTTP/SSE interface", () => { service.dispose(); }); + it("state-only SSE does not capture, and route disposal releases every active viewer", async () => { + const { service, scheduler } = setup({ thumbnails: false }); + service.addSession("s1"); + const { routes, webServer } = routeHarness(); + const dispose = registerObservationRoutes({ get: () => webServer } as never, service); + const route = routes.get("/bsk-observation/events")!; + const metadata = fakeRes(); + await route.handler(fakeReq({ url: "/bsk-observation/events?thumbnails=0" }), metadata.res); + expect(scheduler.pending()).toEqual([]); + const viewing = fakeRes(); + await route.handler(fakeReq({ url: "/bsk-observation/events?thumbnails=1" }), viewing.res); + expect(scheduler.pending()).toEqual([0]); + const endMetadata = vi.spyOn(metadata.res, "end"); + const endViewing = vi.spyOn(viewing.res, "end"); + dispose(); + expect(endMetadata).toHaveBeenCalledOnce(); + expect(endViewing).toHaveBeenCalledOnce(); + expect(scheduler.pending()).toEqual([]); + viewing.res.close(); + metadata.res.close(); + expect(endViewing).toHaveBeenCalledOnce(); + service.dispose(); + }); + + it("releases a viewer when writing its initial snapshot fails", async () => { + const { service, scheduler } = setup({ thumbnails: false }); + service.addSession("s1"); + const { routes, webServer } = routeHarness(); + const dispose = registerObservationRoutes({ get: () => webServer } as never, service); + const response = fakeRes(); + vi.spyOn(response.res, "write").mockImplementation(() => { + throw new Error("closed socket"); + }); + const end = vi.spyOn(response.res, "end"); + await routes.get("/bsk-observation/events")!.handler(fakeReq(), response.res); + expect(end).toHaveBeenCalledOnce(); + expect(scheduler.pending()).toEqual([]); + dispose(); + service.dispose(); + }); + it("refuses a stop without a sessionId", async () => { const registry = new SessionRegistry(5); own(registry, "s1");