fix(windows): preserve pipe continuity and verify process liveness

fix(dsh): synchronize observation state and manage thumbnail capture

.github/workflows/ci.yml

@@ -37,6 +37,21 @@ jobs:
       - name: Run Rust tests
         run: cargo test --workspace --locked
 
+  windows-daemon:
+    name: Windows daemon regression tests
+    runs-on: windows-latest
+
+    steps:
+      - uses: actions/checkout@v6
+
+      - uses: dtolnay/rust-toolchain@stable
+
+      - name: Run Windows named-pipe tests
+        run: cargo test -p bsk --lib --locked daemon::ipc::windows::tests
+
+      - name: Run Windows process liveness tests
+        run: cargo test -p bsk --lib --locked daemon::lockfile::tests
+
   frontend:
     name: Frontend lint, typecheck, tests, build
     runs-on: ubuntu-latest

apps/extension/src/tools/__tests__/dispatcher.test.ts

@@ -49,6 +49,47 @@ describe("ToolDispatcher", () => {
     vi.unstubAllGlobals();
   });
 
+  it.each([
+    "tool.snapshot",
+    "tool.evaluate",
+    "tool.get_html",
+    "tool.console",
+  ])("annotates Chrome extension-access failures from %s", async (method) => {
+    const tab = { id: 7, windowId: 4242, active: true, url: "https://example.test" };
+    vi.stubGlobal("chrome", {
+      tabs: {
+        get: vi.fn(async () => tab),
+        query: vi.fn(async () => [tab]),
+      },
+    });
+    const { transport, sent, deliver } = fakeTransport();
+    const sessions = new SessionManager({
+      agentWindow: {
+        create: vi.fn(async () => 4242),
+        remove: vi.fn(),
+        ensureActiveTab: vi.fn(),
+      },
+    });
+    await sessions.start("test");
+    const error = new Error("Cannot access a chrome-extension:// URL of different extension");
+    const cdp = {
+      send: vi.fn().mockRejectedValue(error),
+      ensureConsoleCapture: vi.fn().mockRejectedValue(error),
+      consoleEntriesSince: vi.fn(),
+    } as unknown as TestDispatcherCdp;
+    const dispatcher = new ToolDispatcher({ transport, sessions, cdp });
+    dispatcher.start();
+    deliver(makeRequest(method, { session_id: "test", expression: "document.title" }));
+    await vi.waitFor(() => expect(sent).toHaveLength(1));
+    expect(sent[0]).toMatchObject({
+      error: {
+        code: "cdp_failed",
+        data: { reason: "cdp_extension_access_denied" },
+      },
+    });
+    dispatcher.stop();
+  });
+
   it("uses the configured recording runtime for start and stop", async () => {
     const { transport, sent, deliver } = fakeTransport();
     const sessions = new SessionManager({

apps/extension/src/tools/__tests__/errors.test.ts

@@ -0,0 +1,39 @@
+import { describe, expect, it } from "vitest";
+import type { RpcError } from "@/transport/types";
+import { cdpError, classifyCdpError } from "../errors";
+
+const denied = "Cannot access a chrome-extension:// URL of different extension";
+
+describe("CDP extension access errors", () => {
+  it("classifies both Chrome Error instances and string failures", () => {
+    for (const error of [new Error(denied), denied]) {
+      expect(cdpError(error)).toEqual({
+        code: "cdp_failed",
+        message: denied,
+        data: { reason: "cdp_extension_access_denied" },
+      });
+    }
+  });
+
+  it("preserves transfer outcome and cleanup instructions when CDP was denied", () => {
+    const error: RpcError = {
+      code: "cdp_failed",
+      message: denied,
+      data: {
+        reason: "transfer_outcome_unknown",
+        effect_state: "unknown",
+        cleanup_state: "failed",
+      },
+    };
+    expect(classifyCdpError(error)).toBe(error);
+  });
+
+  it("leaves other error codes and unrelated debugger failures unchanged", () => {
+    for (const error of [
+      { code: "permission_denied", message: denied },
+      { code: "cdp_failed", message: "Another debugger is already attached" },
+    ] satisfies RpcError[]) {
+      expect(classifyCdpError(error)).toBe(error);
+    }
+  });
+});

apps/extension/src/tools/__tests__/navigation-recovery.test.ts

@@ -0,0 +1,266 @@
+import { describe, expect, it, vi } from "vitest";
+import { SessionManager } from "@/session-manager/manager";
+import type { BrowserNavigationApi } from "../browser-navigation";
+import { handleNavigate, handleReload } from "../navigation";
+import type { CdpRunner } from "../shared";
+
+const denied = "Cannot access a chrome-extension:// URL of different extension";
+const url = "https://example.test/recovered";
+
+function event<T extends (...args: never[]) => void>() {
+  const listeners = new Set<T>();
+  return {
+    listeners,
+    addListener: (listener: T) => listeners.add(listener),
+    removeListener: (listener: T) => listeners.delete(listener),
+    fire: (...args: Parameters<T>) => {
+      for (const listener of [...listeners]) listener(...args);
+    },
+  };
+}
+
+async function fixture() {
+  const manager = new SessionManager({
+    agentWindow: {
+      create: vi.fn(async () => 100),
+      remove: vi.fn(async () => {}),
+      ensureActiveTab: vi.fn(async () => 4),
+    },
+  });
+  await manager.start("test");
+  type Listener = Parameters<BrowserNavigationApi["onCommitted"]["addListener"]>[0];
+  const events = {
+    onCommitted: event<Listener>(),
+    onDOMContentLoaded: event<Listener>(),
+    onCompleted: event<Listener>(),
+    onErrorOccurred: event<Listener>(),
+  };
+  const cdpEvents = event<Parameters<NonNullable<CdpRunner["onEvent"]>>[0]>();
+  const state = { blocked: true, destinationBlocked: false, emitIdle: true };
+  const send = vi.fn(async (_tabId: number, method: string) => {
+    if (state.blocked) throw new Error(denied);
+    if (method === "Page.getFrameTree") {
+      return { frameTree: { frame: { id: "main", loaderId: "new-loader" } } };
+    }
+    if (method === "Runtime.evaluate") return { result: { value: "complete" } };
+    if (method === "Page.setLifecycleEventsEnabled" && state.emitIdle) {
+      cdpEvents.fire({ tabId: 4 }, "Page.lifecycleEvent", {
+        frameId: "main",
+        loaderId: "new-loader",
+        name: "networkIdle",
+      });
+    }
+    if (method === "Page.navigate") return { frameId: "main", loaderId: "new-loader" };
+    return {};
+  });
+  const cdp: CdpRunner = {
+    send: send as CdpRunner["send"],
+    onEvent: (listener) => {
+      cdpEvents.addListener(listener);
+      return { dispose: () => cdpEvents.removeListener(listener) };
+    },
+  };
+  const main = { tabId: 4, frameId: 0, documentId: "new-document" };
+  const navigate = async () => {
+    state.blocked = state.destinationBlocked;
+    events.onCommitted.fire(main);
+    events.onDOMContentLoaded.fire(main);
+    events.onCompleted.fire(main);
+  };
+  const browserNavigation = { ...events, update: vi.fn(navigate), reload: vi.fn(navigate) };
+  const tab = { id: 4, windowId: 100, active: true, url } as chrome.tabs.Tab;
+  const tabsApi = { get: vi.fn(async () => tab), query: vi.fn(async () => [tab]) };
+  const deps = { cdp, tabsApi, browserNavigation, defaultTimeoutMs: 1000 };
+  const expectCleanedUp = () => {
+    for (const entry of Object.values(events)) expect(entry.listeners.size).toBe(0);
+    expect(cdpEvents.listeners.size).toBe(0);
+  };
+  return { manager, deps, state, send, events, main, cdpEvents, expectCleanedUp };
+}
+
+describe("navigation after Chrome denies extension-frame access", () => {
+  it.each([
+    "load",
+    "commit",
+    "domcontentloaded",
+    "networkidle",
+  ] as const)("recovers the same tab with the requested %s checkpoint", async (waitUntil) => {
+    const f = await fixture();
+    const result = await handleNavigate(
+      f.manager,
+      {
+        session_id: "test",
+        url,
+        wait_until: waitUntil,
+      },
+      f.deps,
+    );
+
+    expect(result).toMatchObject({ tab_id: 4, final_url: url, reached: waitUntil });
+    expect(f.deps.browserNavigation.update).toHaveBeenCalledExactlyOnceWith(4, { url });
+    expect(f.send.mock.calls.some(([, method]) => method === "Page.navigate")).toBe(false);
+    expect(f.send).toHaveBeenCalledWith(4, "Page.enable", {});
+    f.expectCleanedUp();
+  });
+
+  it("recovers reload through tabs.reload and preserves hard reload", async () => {
+    const f = await fixture();
+    const result = await handleReload(f.manager, { session_id: "test", hard: true }, f.deps);
+    expect(result).toMatchObject({ tab_id: 4, reached: "load" });
+    expect(f.deps.browserNavigation.reload).toHaveBeenCalledExactlyOnceWith(4, {
+      bypassCache: true,
+    });
+    expect(f.deps.browserNavigation.update).not.toHaveBeenCalled();
+    f.expectCleanedUp();
+  });
+
+  it("reports the restriction when reload leaves the conflicting frame present", async () => {
+    const f = await fixture();
+    f.state.destinationBlocked = true;
+    const result = await handleReload(f.manager, { session_id: "test" }, f.deps);
+    expect(result).toMatchObject({
+      code: "cdp_failed",
+      data: { reason: "cdp_extension_access_denied" },
+    });
+    expect(f.deps.browserNavigation.reload).toHaveBeenCalledOnce();
+    f.expectCleanedUp();
+  });
+
+  it("does not fall back for unrelated debugger failures", async () => {
+    const f = await fixture();
+    f.send.mockRejectedValue(new Error("Another debugger is already attached"));
+    const result = await handleNavigate(f.manager, { session_id: "test", url }, f.deps);
+    expect(result).toMatchObject({
+      code: "cdp_failed",
+      message: "Another debugger is already attached",
+    });
+    expect(f.deps.browserNavigation.update).not.toHaveBeenCalled();
+    f.expectCleanedUp();
+  });
+
+  it("does not replay navigation when a dispatched CDP command fails", async () => {
+    const f = await fixture();
+    f.state.blocked = false;
+    const send = f.send.getMockImplementation()!;
+    f.send.mockImplementation(async (tabId, method) => {
+      if (method === "Page.navigate") throw new Error(denied);
+      return send(tabId, method);
+    });
+    const result = await handleNavigate(f.manager, { session_id: "test", url }, f.deps);
+    expect(result).toMatchObject({
+      code: "cdp_failed",
+      data: { reason: "cdp_extension_access_denied" },
+    });
+    expect(f.deps.browserNavigation.update).not.toHaveBeenCalled();
+    f.expectCleanedUp();
+  });
+
+  it("keeps the Agent Window guard in front of the recovery", async () => {
+    const f = await fixture();
+    f.deps.tabsApi.get.mockResolvedValue({
+      id: 4,
+      windowId: 200,
+      active: true,
+      url,
+    } as chrome.tabs.Tab);
+    const result = await handleNavigate(f.manager, { session_id: "test", tab_id: 4, url }, f.deps);
+    expect(result).toMatchObject({ code: "permission_denied" });
+    expect(f.send).not.toHaveBeenCalled();
+    expect(f.deps.browserNavigation.update).not.toHaveBeenCalled();
+  });
+
+  it("ignores old-document, subframe, and other-tab completion events", async () => {
+    const f = await fixture();
+    f.deps.browserNavigation.update.mockImplementation(async () => {});
+    let settled = false;
+    const work = handleNavigate(f.manager, { session_id: "test", url }, f.deps).then((result) => {
+      settled = true;
+      return result;
+    });
+    await vi.waitFor(() => expect(f.deps.browserNavigation.update).toHaveBeenCalledOnce());
+    f.events.onCompleted.fire(f.main);
+    f.events.onCommitted.fire({ ...f.main, tabId: 9 });
+    f.events.onCommitted.fire({ ...f.main, frameId: 1 });
+    f.events.onCompleted.fire(f.main);
+    await Promise.resolve();
+    expect(settled).toBe(false);
+    f.events.onCommitted.fire(f.main);
+    f.events.onCompleted.fire({ ...f.main, documentId: "old-document" });
+    await Promise.resolve();
+    expect(settled).toBe(false);
+    f.state.blocked = false;
+    f.events.onCompleted.fire(f.main);
+    expect(await work).toMatchObject({ reached: "load" });
+    f.expectCleanedUp();
+  });
+
+  it("does not infer network idle from a browser load event", async () => {
+    const f = await fixture();
+    f.state.emitIdle = false;
+    const result = await handleNavigate(
+      f.manager,
+      {
+        session_id: "test",
+        url,
+        wait_until: "networkidle",
+        timeout_ms: 10,
+      },
+      f.deps,
+    );
+    expect(result).toMatchObject({ reached: "timeout" });
+    f.expectCleanedUp();
+  });
+
+  it("cleans up browser listeners on timeout", async () => {
+    const f = await fixture();
+    f.deps.browserNavigation.update.mockImplementation(async () => {});
+    const result = await handleNavigate(
+      f.manager,
+      { session_id: "test", url, timeout_ms: 10 },
+      f.deps,
+    );
+    expect(result).toMatchObject({ reached: "timeout" });
+    f.expectCleanedUp();
+  });
+
+  it("does not navigate an already-cancelled request", async () => {
+    const f = await fixture();
+    const result = await handleNavigate(
+      f.manager,
+      { session_id: "test", url },
+      {
+        ...f.deps,
+        signal: AbortSignal.abort(),
+      },
+    );
+    expect(result).toMatchObject({ code: "cancelled" });
+    expect(f.deps.browserNavigation.update).not.toHaveBeenCalled();
+    f.expectCleanedUp();
+  });
+
+  it("cleans up browser listeners when cancelled after dispatch", async () => {
+    const f = await fixture();
+    const ac = new AbortController();
+    f.deps.browserNavigation.update.mockImplementation(async () => {
+      ac.abort();
+    });
+    const result = await handleNavigate(
+      f.manager,
+      { session_id: "test", url },
+      { ...f.deps, signal: ac.signal },
+    );
+    expect(result).toMatchObject({ code: "cancelled" });
+    expect(f.deps.browserNavigation.update).toHaveBeenCalledOnce();
+    f.expectCleanedUp();
+  });
+
+  it("reports browser navigation errors and releases listeners", async () => {
+    const f = await fixture();
+    f.deps.browserNavigation.update.mockImplementation(async () => {
+      f.events.onErrorOccurred.fire({ ...f.main, error: "net::ERR_NAME_NOT_RESOLVED" });
+    });
+    const result = await handleNavigate(f.manager, { session_id: "test", url }, f.deps);
+    expect(result).toMatchObject({ code: "cdp_failed", message: "net::ERR_NAME_NOT_RESOLVED" });
+    f.expectCleanedUp();
+  });
+});

apps/extension/src/tools/browser-navigation.ts

@@ -0,0 +1,110 @@
+import type { RpcError, WaitUntil } from "@/transport/types";
+import { cdpError } from "./errors";
+
+interface NavigationEvent {
+  tabId: number;
+  frameId: number;
+  documentId?: string;
+  error?: string;
+}
+
+interface NavigationEvents {
+  addListener(listener: (details: NavigationEvent) => void): void;
+  removeListener(listener: (details: NavigationEvent) => void): void;
+}
+
+/** Normal browser navigation remains available when chrome.debugger is denied. */
+export interface BrowserNavigationApi {
+  update(tabId: number, props: { url: string }): Promise<unknown>;
+  reload(tabId: number, props: { bypassCache: boolean }): Promise<void>;
+  onCommitted: NavigationEvents;
+  onDOMContentLoaded: NavigationEvents;
+  onCompleted: NavigationEvents;
+  onErrorOccurred: NavigationEvents;
+}
+
+export const chromeBrowserNavigationApi: BrowserNavigationApi = {
+  update: (tabId, props) => chrome.tabs.update(tabId, props),
+  reload: (tabId, props) => chrome.tabs.reload(tabId, props),
+  get onCommitted() {
+    return chrome.webNavigation.onCommitted;
+  },
+  get onDOMContentLoaded() {
+    return chrome.webNavigation.onDOMContentLoaded;
+  },
+  get onCompleted() {
+    return chrome.webNavigation.onCompleted;
+  },
+  get onErrorOccurred() {
+    return chrome.webNavigation.onErrorOccurred;
+  },
+};
+
+export type NavigationOutcome =
+  | { reached: "match" | "timeout" | "cancelled"; lastLifecycle?: string }
+  | { reached: "failed"; error: RpcError };
+
+/** Subscribe before dispatch and ignore the departing document's load events. */
+export async function navigateWithBrowserApi(
+  api: BrowserNavigationApi,
+  tabId: number,
+  action: () => Promise<unknown>,
+  waitUntil: Exclude<WaitUntil, "networkidle">,
+  timeoutMs: number,
+  signal?: AbortSignal,
+): Promise<NavigationOutcome> {
+  if (signal?.aborted) return { reached: "cancelled" };
+  if (timeoutMs <= 0) return { reached: "timeout" };
+  let committed = false;
+  let documentId: string | undefined;
+  let lastLifecycle: string | undefined;
+  let finish!: (outcome: NavigationOutcome) => void;
+  const outcome = new Promise<NavigationOutcome>((resolve) => {
+    finish = resolve;
+  });
+  const isMainFrame = (details: NavigationEvent) =>
+    details.tabId === tabId && details.frameId === 0;
+  const onCommitted = (details: NavigationEvent) => {
+    if (!isMainFrame(details)) return;
+    committed = true;
+    documentId = details.documentId;
+    lastLifecycle = "commit";
+    if (waitUntil === "commit") finish({ reached: "match", lastLifecycle });
+  };
+  const onDOMContentLoaded = (details: NavigationEvent) => {
+    if (!isMainFrame(details) || !committed || (documentId && details.documentId !== documentId))
+      return;
+    lastLifecycle = "DOMContentLoaded";
+    if (waitUntil === "domcontentloaded") finish({ reached: "match", lastLifecycle });
+  };
+  const onCompleted = (details: NavigationEvent) => {
+    if (!isMainFrame(details) || !committed || (documentId && details.documentId !== documentId))
+      return;
+    finish({ reached: "match", lastLifecycle: "load" });
+  };
+  const onError = (details: NavigationEvent) => {
+    if (!isMainFrame(details)) return;
+    finish({ reached: "failed", error: cdpError(details.error ?? "browser navigation failed") });
+  };
+  const onAbort = () => finish({ reached: "cancelled", lastLifecycle });
+  const subscriptions = [
+    [api.onCommitted, onCommitted],
+    [api.onDOMContentLoaded, onDOMContentLoaded],
+    [api.onCompleted, onCompleted],
+    [api.onErrorOccurred, onError],
+  ] as const;
+  for (const [event, listener] of subscriptions) event.addListener(listener);
+  signal?.addEventListener("abort", onAbort, { once: true });
+  const timer = setTimeout(() => finish({ reached: "timeout", lastLifecycle }), timeoutMs);
+  try {
+    if (signal?.aborted) return { reached: "cancelled" };
+    await action();
+    return await outcome;
+  } catch (error) {
+    return { reached: "failed", error: cdpError(error) };
+  } finally {
+    clearTimeout(timer);
+    signal?.removeEventListener("abort", onAbort);
+    for (const [event, listener] of subscriptions) event.removeListener(listener);
+  }
+}

apps/extension/src/tools/dispatcher.ts

@@ -36,6 +36,7 @@ import { isRequestFrame } from "@/transport/types";
 import { handleConsole } from "./console";
 import { handleDownload } from "./download";
 import { type EmulateCdpRunner, handleEmulate } from "./emulate";
+import { classifyCdpError } from "./errors";
 import { handleEvaluate } from "./evaluate";
 import { handleRequestHelp } from "./human-loop";
 import { handleClick, handleFill, handleHover, handlePress, handleSelect } from "./interaction";
@@ -239,7 +240,7 @@ export class ToolDispatcher {
       if (sessionId) this.onBrowserControlResumed?.(sessionId);
       const result = await this.invoke(req, ac.signal);
       if (isRpcError(result)) {
-        body = { id: req.id, error: result };
+        body = { id: req.id, error: classifyCdpError(result) };
       } else {
         body = { id: req.id, result };
         if (req.method === "tool.session_start") {

apps/extension/src/tools/errors.ts

@@ -13,6 +13,34 @@ import type {
 
 export type { RpcErrorData, RpcErrorReason };
 
+/** Chrome checks the whole tab's frame tree, including other extensions' frames. */
+export function isCdpExtensionAccessDenied(error: unknown): boolean {
+  const message = error instanceof Error ? error.message : String(error);
+  return message.includes("Cannot access a chrome-extension:// URL of different extension");
+}
+
+/** Preserve more specific reasons, especially file-transfer outcome/cleanup details. */
+export function classifyCdpError(error: RpcError): RpcError {
+  if (
+    error.code !== "cdp_failed" ||
+    error.data?.reason ||
+    !isCdpExtensionAccessDenied(error.message)
+  ) {
+    return error;
+  }
+  return {
+    ...error,
+    data: { ...error.data, reason: "cdp_extension_access_denied" },
+  };
+}
+
+export function cdpError(error: unknown): RpcError {
+  return classifyCdpError({
+    code: "cdp_failed",
+    message: error instanceof Error ? error.message : String(error),
+  });
+}
+
 export interface TransferErrorOptions {
   effectState: TransferEffectState;
   phase: string;

apps/extension/src/tools/navigation.ts

@@ -28,7 +28,13 @@ import type {
   RpcError,
   WaitUntil,
 } from "@/transport/types";
+import {
+  type BrowserNavigationApi,
+  chromeBrowserNavigationApi,
+  navigateWithBrowserApi,
+} from "./browser-navigation";
 import { attachDialogs, markDialogCursor } from "./dialogs";
+import { cdpError, isCdpExtensionAccessDenied } from "./errors";
 import {
   type CdpRunner,
   type ChromeTabsApi,
@@ -42,6 +48,7 @@ import {
 export interface NavigationDeps {
   cdp: CdpRunner;
   tabsApi: ChromeTabsApi;
+  browserNavigation?: BrowserNavigationApi;
   /** Optional AbortSignal — M7 abort hook (M10.2 will wire the full chain). */
   signal?: AbortSignal;
   /** Override default timeout when the caller omits `timeout_ms`. */
@@ -479,6 +486,81 @@ async function readTabUrl(api: ChromeTabsApi, tabId: number): Promise<string | u
   }
 }
 
+/** Recover only a failed preflight; never replay an already-dispatched navigation. */
+async function recoverBrowserNavigation(
+  deps: NavigationDeps,
+  tabId: number,
+  action: (api: BrowserNavigationApi) => Promise<unknown>,
+  waitUntil: WaitUntil,
+  timeoutMs: number,
+): Promise<ReloadResult | RpcError> {
+  const abort = linkedAbortSignal(deps.signal);
+  const deadline = Date.now() + timeoutMs;
+  const api = deps.browserNavigation ?? chromeBrowserNavigationApi;
+  try {
+    let outcome = await navigateWithBrowserApi(
+      api,
+      tabId,
+      () => action(api),
+      waitUntil === "networkidle" ? "commit" : waitUntil,
+      timeoutMs,
+      abort.signal,
+    );
+    if (outcome.reached === "failed") return outcome.error;
+    if (outcome.reached === "cancelled" || abort.signal.aborted) {
+      return { code: "cancelled", message: "navigation recovery aborted" };
+    }
+    if (outcome.reached === "match") {
+      // A completed reload is not a recovery if the restricted frame remains.
+      await deps.cdp.send(tabId, "Page.enable", {});
+      if (abort.signal.aborted)
+        return { code: "cancelled", message: "navigation recovery aborted" };
+      if (waitUntil === "networkidle") {
+        // webNavigation cannot prove network idle. Subscribe on the new
+        // document before enabling CDP lifecycle events after its commit.
+        const frame = await readMainFrameInfo(deps.cdp, tabId);
+        if (!frame.frameId) return cdpError("no main frame after navigation");
+        const remaining = deadline - Date.now();
+        if (remaining <= 0) {
+          outcome = { reached: "timeout", lastLifecycle: "commit" };
+        } else {
+          const wait = startLifecycleWait(
+            deps.cdp,
+            tabId,
+            frame.frameId,
+            "networkIdle",
+            remaining,
+            abort.signal,
+            { loaderId: frame.loaderId ?? undefined },
+          );
+          await deps.cdp.send(tabId, "Page.setLifecycleEventsEnabled", { enabled: true });
+          outcome = await wait.promise;
+        }
+      }
+    }
+    if (outcome.reached === "cancelled" || abort.signal.aborted) {
+      return { code: "cancelled", message: "navigation recovery aborted" };
+    }
+    return {
+      tab_id: tabId,
+      final_url: await readTabUrl(deps.tabsApi, tabId),
+      reached: outcome.reached === "match" ? waitUntil : "timeout",
+      ...(outcome.reached === "timeout"
+        ? {
+            error_text: `timed out waiting for ${waitUntil} during navigation recovery after ${timeoutMs}ms`,
+          }
+        : {}),
+    };
+  } catch (error) {
+    return abort.signal.aborted
+      ? { code: "cancelled", message: "navigation recovery aborted" }
+      : cdpError(error);
+  } finally {
+    abort.abort();
+    abort.cleanup();
+  }
+}
+
 // ---------------------------------------------------------------------------
 // tool.navigate
 // ---------------------------------------------------------------------------
@@ -505,7 +587,20 @@ export async function handleNavigate(
 
   try {
     deps.cdp.trackSessionTab?.(ctx.sessionId, target.tabId);
-    await ensureCdpReady(deps.cdp, target.tabId);
+    try {
+      await ensureCdpReady(deps.cdp, target.tabId);
+    } catch (error) {
+      if (!isCdpExtensionAccessDenied(error)) throw error;
+      const recovered = await recoverBrowserNavigation(
+        deps,
+        target.tabId,
+        (api) => api.update(target.tabId, { url: params.url }),
+        waitUntil,
+        timeoutMs,
+      );
+      if (isRpcError(recovered)) return recovered;
+      return attachDialogs(deps.cdp, target.tabId, dialogCursor, { ...recovered, url: params.url });
+    }
     const expected = cdpLifecycleName(waitUntil);
     const beforeReadyState = await probeMainFrameReadyState(deps.cdp, target.tabId);
     const beforeFrame = await readMainFrameInfo(deps.cdp, target.tabId);
@@ -574,10 +669,7 @@ export async function handleNavigate(
       }`,
     });
   } catch (err) {
-    return {
-      code: "cdp_failed",
-      message: err instanceof Error ? err.message : String(err),
-    };
+    return cdpError(err);
   }
 }
 
@@ -727,7 +819,24 @@ export async function handleReload(
 
   try {
     deps.cdp.trackSessionTab?.(ctx.sessionId, target.tabId);
-    await ensureCdpReady(deps.cdp, target.tabId);
+    try {
+      await ensureCdpReady(deps.cdp, target.tabId);
+    } catch (error) {
+      if (!isCdpExtensionAccessDenied(error)) throw error;
+      const previousUrl = await readTabUrl(deps.tabsApi, target.tabId);
+      const recovered = await recoverBrowserNavigation(
+        deps,
+        target.tabId,
+        (api) => api.reload(target.tabId, { bypassCache: ignoreCache }),
+        waitUntil,
+        timeoutMs,
+      );
+      if (isRpcError(recovered)) return recovered;
+      return attachDialogs(deps.cdp, target.tabId, dialogCursor, {
+        ...recovered,
+        previous_url: previousUrl,
+      });
+    }
     const previousUrl = await readTabUrl(deps.tabsApi, target.tabId);
     const beforeFrame = await readMainFrameInfo(deps.cdp, target.tabId);
     const expected = cdpLifecycleName(waitUntil);
@@ -776,10 +885,7 @@ export async function handleReload(
       }`,
     });
   } catch (err) {
-    return {
-      code: "cdp_failed",
-      message: err instanceof Error ? err.message : String(err),
-    };
+    return cdpError(err);
   }
 }
 

apps/extension/src/transport/types.ts

@@ -36,6 +36,7 @@ export type RpcErrorReason =
   | "single_select_value_count"
   | "tab_not_active"
   | "restricted_tab_url"
+  | "cdp_extension_access_denied"
   | "borrow_conflict"
   | "screenshot_capture_failed"
   | "file_input_probe_failed"

crates/bsk-cli/src/cli/render_error.rs

@@ -52,6 +52,7 @@ pub mod reason {
     pub const OPTION_NOT_FOUND: &str = "option_not_found";
     pub const SINGLE_SELECT_VALUE_COUNT: &str = "single_select_value_count";
     pub const TAB_NOT_ACTIVE: &str = "tab_not_active";
+    pub const CDP_EXTENSION_ACCESS_DENIED: &str = "cdp_extension_access_denied";
     pub const BORROW_CONFLICT: &str = "borrow_conflict";
     pub const SCREENSHOT_CAPTURE_FAILED: &str = "screenshot_capture_failed";
     pub const FILE_INPUT_PROBE_FAILED: &str = "file_input_probe_failed";
@@ -211,6 +212,13 @@ pub fn info_for_error(code: ErrorCode, data: Option<&serde_json::Value>) -> Rend
         return base;
     };
     match (code, reason) {
+        (ErrorCode::CdpFailed, reason::CDP_EXTENSION_ACCESS_DENIED) => RenderInfo {
+            summary: "Chrome blocked CDP access to another extension's content in this tab",
+            hint: Some(
+                "a web page can contain a restricted extension frame; disable the conflicting extension and reload, or use `bsk navigate <url>` to leave this page; reconnecting BrowserSkill alone does not remove the restriction",
+            ),
+            exit_code: base.exit_code,
+        },
         (_, reason::TRANSFER_OUTCOME_UNKNOWN) => RenderInfo {
             summary: "the file transfer outcome could not be confirmed",
             hint: Some(
@@ -544,6 +552,18 @@ mod tests {
         assert!(!info.summary.contains("sandbox"));
     }
 
+    #[test]
+    fn extension_access_denied_explains_the_frame_restriction_and_navigation_recovery() {
+        let data = serde_json::json!({ "reason": reason::CDP_EXTENSION_ACCESS_DENIED });
+        let info = info_for_error(ErrorCode::CdpFailed, Some(&data));
+        assert!(info.summary.contains("another extension"));
+        let hint = info.hint.unwrap();
+        assert!(hint.contains("restricted extension frame"));
+        assert!(hint.contains("bsk navigate <url>"));
+        assert!(hint.contains("disable the conflicting extension"));
+        assert_eq!(info.exit_code, 3);
+    }
+
     #[test]
     fn session_busy_overrides_timeout_copy() {
         let data = serde_json::json!({ "reason": reason::SESSION_BUSY });

crates/bsk-cli/src/daemon/ipc.rs

@@ -1399,6 +1399,28 @@ mod windows {
                 connected = pipe.connect() => {
                     match connected {
                         Ok(()) => {
+                            // Keep the connected instance alive until its replacement
+                            // exists. Otherwise a fast handler can close the last
+                            // instance and make new clients fail with NotFound.
+                            loop {
+                                match ServerOptions::new()
+                                    .access_inbound(true)
+                                    .access_outbound(true)
+                                    .create(&listener.pipe_name)
+                                {
+                                    Ok(next) => {
+                                        listener.first = Some(next);
+                                        break;
+                                    }
+                                    Err(err) => {
+                                        warn!(?err, "create next named-pipe instance failed");
+                                        tokio::select! {
+                                            _ = &mut shutdown => return,
+                                            _ = tokio::time::sleep(std::time::Duration::from_millis(50)) => {}
+                                        }
+                                    }
+                                }
+                            }
                             on_open();
                             let handler = handler.clone();
                             let on_act = on_activity.clone();
@@ -1467,6 +1489,118 @@ mod windows {
         }
         Ok(())
     }
+    #[cfg(test)]
+    mod tests {
+        use super::*;
+        use std::sync::Mutex;
+        use std::time::Duration;
+        use tokio::net::windows::named_pipe::ClientOptions;
+        use tokio::sync::oneshot;
+
+        fn isolated_listener() -> NamedPipeListener {
+            let pipe_name = format!(r"\\.\pipe\bsk-test-{}", uuid::Uuid::new_v4());
+            let first = ServerOptions::new()
+                .first_pipe_instance(true)
+                .create(&pipe_name)
+                .unwrap();
+            NamedPipeListener {
+                pipe_name,
+                first: Some(first),
+            }
+        }
+
+        #[tokio::test]
+        async fn next_instance_exists_before_connection_is_handed_off() {
+            let listener = isolated_listener();
+            let name = listener.pipe_name.clone();
+            let probe_name = name.clone();
+            let (probe_tx, probe_rx) = oneshot::channel();
+            let probe_tx = Mutex::new(Some(probe_tx));
+            let (stop_tx, stop_rx) = oneshot::channel();
+            let server = tokio::spawn(serve(
+                listener,
+                crate::daemon::ipc::default_ping_handler(),
+                move || {
+                    if let Some(tx) = probe_tx.lock().unwrap().take() {
+                        // This callback runs before the connection task can finish.
+                        // A second client must already have an instance to open.
+                        let result = ClientOptions::new().open(&probe_name).map(drop);
+                        let _ = tx.send(result);
+                    }
+                },
+                || {},
+                || {},
+                async {
+                    let _ = stop_rx.await;
+                },
+            ));
+            let client = ClientOptions::new().open(&name).unwrap();
+            let result = tokio::time::timeout(Duration::from_secs(5), probe_rx).await;
+            drop(client);
+            let _ = stop_tx.send(());
+            server.await.unwrap();
+            result
+                .expect("accept callback ran")
+                .unwrap()
+                .expect("next pipe instance is ready");
+        }
+
+        #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+        async fn rapid_disconnects_and_concurrent_rpc_connections() {
+            let listener = isolated_listener();
+            let name = listener.pipe_name.clone();
+            let (stop_tx, stop_rx) = oneshot::channel();
+            let server = tokio::spawn(serve(
+                listener,
+                crate::daemon::ipc::default_ping_handler(),
+                || {},
+                || {},
+                || {},
+                async {
+                    let _ = stop_rx.await;
+                },
+            ));
+            let exercise = async {
+                // Include clients that close without sending any request.
+                for _ in 0..64 {
+                    drop(
+                        crate::ipc_client::Client::connect_path(name.clone().into())
+                            .await
+                            .unwrap(),
+                    );
+                }
+                let mut clients = tokio::task::JoinSet::new();
+                for _ in 0..8 {
+                    let name = name.clone();
+                    clients.spawn(async move {
+                        for _ in 0..32 {
+                            let mut client =
+                                crate::ipc_client::Client::connect_path(name.clone().into())
+                                    .await
+                                    .unwrap();
+                            let reply: bsk_protocol::PingResult = client
+                                .call(
+                                    bsk_protocol::Method::SystemPing,
+                                    &serde_json::json!({}),
+                                    Duration::from_secs(5),
+                                )
+                                .await
+                                .unwrap()
+                                .unwrap();
+                            assert!(reply.pong);
+                        }
+                    });
+                }
+                while let Some(result) = clients.join_next().await {
+                    result.unwrap();
+                }
+            };
+            let result = tokio::time::timeout(Duration::from_secs(30), exercise).await;
+            let _ = stop_tx.send(());
+            server.await.unwrap();
+            result.expect("connection exercise completed");
+        }
+    }
 }
 
 #[cfg(windows)]

crates/bsk-cli/src/daemon/lockfile.rs

@@ -90,17 +90,21 @@ pub fn pid_alive(pid: u32) -> bool {
 
     #[cfg(windows)]
     {
-        use windows_sys::Win32::Foundation::CloseHandle;
+        use windows_sys::Win32::Foundation::{CloseHandle, WAIT_TIMEOUT};
         use windows_sys::Win32::System::Threading::{
-            OpenProcess, PROCESS_QUERY_LIMITED_INFORMATION,
+            OpenProcess, PROCESS_SYNCHRONIZE, WaitForSingleObject,
         };
         unsafe {
-            let handle = OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, 0, pid);
+            let handle = OpenProcess(PROCESS_SYNCHRONIZE, 0, pid);
             if handle.is_null() {
                 return false;
             }
+            // An exited process can still be opened while another handle
+            // keeps its kernel object alive. Poll its termination signal;
+            // unlike checking STILL_ACTIVE, this also handles exit code 259.
+            let alive = WaitForSingleObject(handle, 0) == WAIT_TIMEOUT;
             CloseHandle(handle);
-            true
+            alive
         }
     }
 
@@ -110,3 +114,39 @@ pub fn pid_alive(pid: u32) -> bool {
         false
     }
 }
+
+#[cfg(all(test, windows))]
+mod tests {
+    use super::pid_alive;
+    use std::os::windows::process::CommandExt;
+    use std::process::Command;
+
+    #[test]
+    fn current_process_is_alive() {
+        assert!(pid_alive(std::process::id()));
+    }
+
+    #[test]
+    fn invalid_process_is_not_alive() {
+        assert!(!pid_alive(0));
+        assert!(!pid_alive(u32::MAX));
+    }
+
+    #[test]
+    fn exited_process_with_retained_handle_is_not_alive() {
+        for exit_code in [0, 259] {
+            let mut child = Command::new("cmd.exe")
+                .args(["/D", "/C", &format!("exit {exit_code}")])
+                .creation_flags(0x0800_0000) // CREATE_NO_WINDOW
+                .spawn()
+                .expect("spawn short-lived process");
+            assert_eq!(child.wait().unwrap().code(), Some(exit_code));
+            // Keep Child (and its process handle) alive during the probe.
+            assert!(
+                !pid_alive(child.id()),
+                "exited process with code {exit_code}"
+            );
+            drop(child);
+        }
+    }
+}

packages/dsh-plugin-browserskill/README.md

@@ -112,7 +112,7 @@ shell's theme cannot bleed back in.
 - **Lifecycle**: hidden while the plugin owns no sessions; appears on the first
   `browser_session` start action; disappears when all sessions stop (or the plugin unloads).
 - **Focus view**: status row (green/idle/red dot + session + action + mm:ss), the latest page
-  frame (refreshes every ~1.5s while active, ~8s when idle; the last good frame stays on
+  frame (while viewed, refreshes every ~1.5s while active, ~8s when idle; the last good frame stays on
   stage while the next one loads, and is kept on errors so the card does not flash),
   and a compact icon toolbar (Interrupt + Pop out, hover for the label).
 - **Interrupt**: one click kills the in-flight bsk command of the focus session (same semantics
@@ -129,11 +129,32 @@ shell's theme cannot bleed back in.
 - **Pop out (PiP)**: upgrades the card into a native Document PiP window (requires a user
   gesture, per browser rules), sized from the current card; closing the PiP falls back to the
   in-page card with state intact. Browsers without Document PiP simply hide the button.
-- **Wire**: the host serves `GET /bsk-observation/state`, `GET /bsk-observation/events` (SSE),
-  `POST /bsk-observation/interrupt`, and `GET /bsk-observation/thumbnail/<attachmentId>` over
-  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.
+- **Screenshot demand**: the client requests periodic screenshots while an observation view is
+  visible, using the PiP document's visibility when popped out. Hidden or collapsed views can
+  keep subscribing to state without requesting screenshots. The first screenshot subscriber
+  starts capture scheduling; the last one's departure cancels timers and skips queued captures.
+  An already running capture may finish. Demand applies to the service's owned sessions, not just
+  the selected session.
+- **Wire**: `GET /bsk-observation/events` is the live synchronization channel (SSE). Every connection,
+  including reconnects, starts with `{ type: "snapshot", sessions, available }`, followed by
+  `upsert`, `remove`, `reset`, and `availability` events on the same stream. Replace local state
+  with each snapshot before applying subsequent events. Use `?thumbnails=0` for state-only
+  subscriptions or `?thumbnails=1` to request screenshots; omitting the parameter defaults to
+  requesting screenshots for compatibility. Closing the stream releases its screenshot demand.
+  `GET /bsk-observation/state` remains a one-time `{ sessions, available }` read and does not
+  request screenshots. It has no ordering guarantee relative to a separate SSE connection;
+  live clients should initialize and resynchronize from the stream's snapshot instead.
+  The host also serves `POST /bsk-observation/interrupt`, `POST /bsk-observation/stop`, and
+  `GET /bsk-observation/thumbnail/<attachmentId>` through 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.
+- **Programmatic subscription**: `ObservationService.subscribe(listener, { thumbnails: false })`
+  subscribes to state without requesting screenshots. On an active service, the listener receives
+  the initial `snapshot` synchronously, before `subscribe` returns, then receives subsequent
+  changes. `thumbnails` defaults to `true`, so callers should choose it explicitly. The returned
+  unsubscribe function is safe to call repeatedly and releases this subscription's screenshot
+  demand. Subscribing after service disposal is a no-op.
 - **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

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;

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 (
-    <div className={cn(css.body, "bsk-obs")} data-state={state} data-in-pip={inPip || undefined}>
+    <div
+      ref={viewRef}
+      className={cn(css.body, "bsk-obs")}
+      data-state={state}
+      data-in-pip={inPip || undefined}
+    >
       <div
         className={css.header}
         data-testid="obs-header"
@@ -421,9 +433,15 @@ export function OverlayBody(props: {
           </div>
         )}
         {thumb?.status === "error" ? (
-          <span className={css.badge} aria-label="thumbnail failed">
+          <button
+            type="button"
+            className={css.badge}
+            aria-label="Retry thumbnail"
+            title="Frame unavailable — retry"
+            onClick={() => store.retryThumbnail(thumbId)}
+          >
             <IconWarn size={12} />
-          </span>
+          </button>
         ) : null}
       </div>
       {sessions.length >= 2 ? (

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<string, ThumbnailState>();
   /** sessionId → the attachment id currently advertised for it. */
   private readonly thumbBySession = new Map<string, string>();
-  /** 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<string, { attachmentId: string; url: string }>();
+  private readonly retryTimers = new Map<string, ReturnType<typeof setTimeout>>();
+  private readonly retryCounts = new Map<string, number>();
   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<boolean> {
     try {

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<HTMLDivElement>(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;

packages/dsh-plugin-browserskill/src/observation-http.ts

@@ -5,12 +5,19 @@
  * the event allowlist lives in dsh-api-remotes), so the plugin serves its
  * observation channel through the documented `webServer` route seam instead:
  *
- *   GET  /bsk-observation/state      → { sessions, available }
- *   GET  /bsk-observation/events     → SSE stream of ObservationEvent
+ *   GET  /bsk-observation/state      → one-time { sessions, available } read
+ *   GET  /bsk-observation/events     → SSE snapshot, then ObservationEvent increments
+ *                                    (on every connection, including reconnects)
+ *                                    ?thumbnails=0: state only; =1: request screenshots
+ *                                    Omitted parameter defaults to screenshots.
  *   POST /bsk-observation/interrupt  → body {sessionId?} → {interrupted: boolean}
  *   POST /bsk-observation/stop       → body {sessionId} → {stopped: boolean}
  *   GET  /bsk-observation/thumbnail/<attachmentId> → image bytes
  *
+ * Live clients initialize/resynchronize from the SSE snapshot; a separate
+ * /state read has no ordering guarantee relative to the stream and does not
+ * request screenshots. Closing an SSE connection releases its screenshot demand.
+ *
  * Routes exist only when a webServer service is mounted (web composition);
  * headless deployments skip registration entirely.
  */
@@ -99,6 +106,7 @@ export function registerObservationRoutes(
   if (webServer === undefined) {
     return () => {};
   }
+  const streams = new Set<() => void>();
   const disposers = [
     webServer.register({
       kind: "exact",
@@ -129,14 +137,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<typeof setInterval> | 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 +264,7 @@ export function registerObservationRoutes(
     }),
   ];
   return () => {
+    for (const close of streams) close();
     for (const dispose of disposers) dispose();
   };
 }

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<string, SessionObservation>();
   private readonly listeners = new Set<(event: ObservationEvent) => void>();
+  private thumbnailViewers = 0;
   private readonly captureTimers = new Map<string, unknown>();
   private readonly captureInFlight = new Set<string>();
   /** Captures intentionally cancelled to make way for foreground work. */
@@ -125,7 +127,7 @@ export class ObservationService {
     return this.deps.scheduler ?? DEFAULT_SCHEDULER;
   }
 
-  /** All current entries (client initial/resync snapshot). */
+  /** One-time state read; use subscribe for an ordered snapshot and subsequent changes. */
   getState(): SessionObservation[] {
     return [...this.observations.values()].map((entry) => ({ ...entry }));
   }
@@ -141,10 +143,42 @@ 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);
+  /**
+   * Subscribe to an ordered initial snapshot and subsequent changes. On an
+   * active service, listener receives the snapshot synchronously before this
+   * method returns. Subscribing after disposal is a no-op.
+   *
+   * `thumbnails` defaults to true for compatibility: this subscription counts
+   * as a screenshot viewer for the service's owned sessions. Pass false for
+   * state-only observation. The first viewer starts capture scheduling.
+   *
+   * @returns An idempotent unsubscribe function releasing this subscription's
+   * screenshot demand. When the last viewer leaves, scheduled/queued captures
+   * are cancelled/skipped; an already running capture may finish.
+   */
+  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 +389,7 @@ export class ObservationService {
     this.observations.clear();
     this.emit({ type: "reset" });
     this.listeners.clear();
+    this.thumbnailViewers = 0;
   }
 
   // ------------------------------------------------------------------
@@ -369,12 +404,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 +443,7 @@ export class ObservationService {
    * Failures keep the previous frame and back off silently.
    */
   private async capture(sessionId: string): Promise<void> {
-    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 +452,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 +487,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)) {

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<typeof vi.fn>;
+  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<typeof vi.fn>;
+    }> = [];
+    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(<ObservationOverlay store={h.store} />);
+    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(<ObservationOverlay store={h.store} />);
+    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(<ObservationOverlay store={h.store} />);
+    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<string, unknown>).documentPictureInPicture = {
+      requestWindow: async () => ({ document: pipDoc, addEventListener: () => {}, close: vi.fn() }),
+    };
+    const h = makeHarness([BUSY]);
+    render(<ObservationOverlay store={h.store} />);
+    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(<ObservationOverlay store={h.store} />);
+    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" }]);

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}`,
   });

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<string, string> };
   }[] = [];
   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<T>() {
+    let resolve!: (value: T) => void;
+    let reject!: (reason: Error) => void;
+    const promise = new Promise<T>((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<string>();
+    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<string>();
+    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);
+  });
+});

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<typeof fakeScheduler>;
+  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<void>((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<void>((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");