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

2026-09-10 13:01:23 +08:00 · 2026-09-07 16:24:04 +08:00
parent 5ebe609c40
commit e961395be9
10 changed files with 800 additions and 122 deletions
@@ -142,6 +142,8 @@
  right: 6px;
  width: 16px;
  height: 16px;
+  padding: 0;
+  cursor: pointer;
  border-radius: 50%;
  display: inline-flex;
  align-items: center;
@@ -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 ? (
@@ -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 {
@@ -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;
@@ -6,7 +6,8 @@
 * observation channel through the documented `webServer` route seam instead:
 *
 *   GET  /bsk-observation/state      → { sessions, available }
- *   GET  /bsk-observation/events     → SSE stream of ObservationEvent
+ *   GET  /bsk-observation/events     → SSE snapshot, then ObservationEvent increments
+ *                                    (?thumbnails=0 for state-only subscribers)
 *   POST /bsk-observation/interrupt  → body {sessionId?} → {interrupted: boolean}
 *   POST /bsk-observation/stop       → body {sessionId} → {stopped: boolean}
 *   GET  /bsk-observation/thumbnail/<attachmentId> → image bytes
@@ -99,6 +100,7 @@ export function registerObservationRoutes(
  if (webServer === undefined) {
    return () => {};
  }
+  const streams = new Set<() => void>();
  const disposers = [
    webServer.register({
      kind: "exact",
@@ -129,14 +131,42 @@ export function registerObservationRoutes(
          "cache-control": "no-cache",
          connection: "keep-alive",
        });
-        const unsubscribe = observation.subscribe((event) => {
-          res.write(`data: ${JSON.stringify(event)}\n\n`);
-        });
-        const heartbeat = setInterval(() => res.write(": heartbeat\n\n"), SSE_HEARTBEAT_MS);
-        res.on("close", () => {
+        const thumbnails =
+          new URL(req.url ?? "/", "http://localhost").searchParams.get("thumbnails") !== "0";
+        let unsubscribe: (() => void) | undefined;
+        let heartbeat: ReturnType<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 +258,7 @@ export function registerObservationRoutes(
    }),
  ];
  return () => {
+    for (const close of streams) close();
    for (const dispose of disposers) dispose();
  };
 }
@@ -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. */
@@ -141,10 +143,30 @@ export class ObservationService {
    this.emit({ type: "availability", available });
  }

-  /** Subscribe to incremental changes; returns an unsubscribe function. */
-  subscribe(listener: (event: ObservationEvent) => void): () => void {
-    this.listeners.add(listener);
-    return () => this.listeners.delete(listener);
+  /** A snapshot and subsequent changes share one ordered subscription. */
+  subscribe(
+    listener: (event: ObservationEvent) => void,
+    { thumbnails = true }: { thumbnails?: boolean } = {},
+  ): () => void {
+    if (this.disposed) return () => {};
+    // Each subscription owns its lease, even if a caller reuses a callback.
+    const receive = (event: ObservationEvent) => listener(event);
+    this.listeners.add(receive);
+    try {
+      listener({ type: "snapshot", sessions: this.getState(), available: this.available });
+    } catch (error) {
+      this.listeners.delete(receive);
+      throw error;
+    }
+    if (thumbnails && ++this.thumbnailViewers === 1) {
+      for (const sessionId of this.observations.keys()) this.scheduleCapture(sessionId, 0);
+    }
+    return () => {
+      if (!this.listeners.delete(receive)) return;
+      if (thumbnails && --this.thumbnailViewers === 0) {
+        for (const sessionId of this.captureTimers.keys()) this.cancelCapture(sessionId);
+      }
+    };
  }

  private emit(event: ObservationEvent): void {
@@ -355,6 +377,7 @@ export class ObservationService {
    this.observations.clear();
    this.emit({ type: "reset" });
    this.listeners.clear();
+    this.thumbnailViewers = 0;
  }

  // ------------------------------------------------------------------
@@ -369,12 +392,21 @@ export class ObservationService {
    }
  }

+  private canCapture(sessionId: string): boolean {
+    const entry = this.observations.get(sessionId);
+    return (
+      this.deps.options.enabled &&
+      !this.disposed &&
+      this.thumbnailViewers > 0 &&
+      !this.foregroundDepth.has(sessionId) &&
+      entry !== undefined &&
+      entry.dead !== true
+    );
+  }
+
  /** Schedule the next capture for a session; `delayMs` 0 means "as soon as the event loop allows". */
  private scheduleCapture(sessionId: string, delayMs?: number): void {
-    if (!this.deps.options.enabled || this.disposed) return;
-    if (this.foregroundDepth.has(sessionId)) return;
-    const entry = this.observations.get(sessionId);
-    if (entry === undefined || entry.dead === true) return;
+    if (!this.canCapture(sessionId)) return;
    this.cancelCapture(sessionId);
    const now = this.scheduler.now();
    const lastSeen = this.lastActivity.get(sessionId) ?? 0;
@@ -399,9 +431,7 @@ export class ObservationService {
   * Failures keep the previous frame and back off silently.
   */
  private async capture(sessionId: string): Promise<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 +440,8 @@ export class ObservationService {
    let writtenPath = outPath;
    try {
      const result = await this.deps.queue.run(sessionId, () => {
-        // A foreground lease may have arrived while this capture was queued.
-        if (this.foregroundDepth.has(sessionId)) return Promise.resolve(undefined);
+        // The last viewer may have left, or foreground work arrived, while queued.
+        if (!this.canCapture(sessionId)) return Promise.resolve(undefined);
        return this.deps.runner.run(["screenshot", "--session", sessionId, "--out", outPath], {
          timeoutMs: 15_000,
          tag: `observation:${sessionId}`,
@@ -445,6 +475,7 @@ export class ObservationService {
      this.setAvailable(true);
      this.publishFrame(sessionId, entry, data);
    } catch {
+      if (this.disposed || !this.observations.has(sessionId)) return;
      // Foreground preemption is healthy scheduling, not browser/daemon
      // unavailability; do not let normal tool traffic trip capture backoff.
      if (!this.capturePreempted.delete(sessionId)) {