fix(dsh-plugin-browserskill): keep observation frames in a 2-slot ring

Stop writing overlay thumbnails into the durable attachment store. Each
session now holds only the current and previous PNG in process memory,
reuses the id when the hash is unchanged, and drops the rest on rotate,
remove, or dispose.

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

@@ -1,16 +1,18 @@
 /**
  * ObservationService: per-owned-session live observation state for the PiP
- * overlay — current action, page url, and a breathing thumbnail carried by
- * attachment reference. Strict ownership boundary applies throughout: only
- * sessions in the plugin's SessionRegistry (owned by construction) ever get
- * an observation entry, and interrupt/kill paths can only reach children this
- * plugin spawned.
+ * overlay — current action, page url, and a breathing thumbnail identified by
+ * an ephemeral frame id. Frames live in a per-session 2-slot ring in process
+ * memory (never the durable attachment store). Strict ownership boundary
+ * applies throughout: only sessions in the plugin's SessionRegistry (owned by
+ * construction) ever get an observation entry, and interrupt/kill paths can
+ * only reach children this plugin spawned.
  *
  * Observation traffic isolation: thumbnail captures run through the runner
  * directly (never through the tool-level instrumentation), emit no action
  * events, and never move the registry's current pointer.
  */
 
+import { createHash } from "node:crypto";
 import { readFile, unlink } from "node:fs/promises";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
@@ -28,7 +30,7 @@ export interface SessionObservation {
   action: string;
   /** Epoch ms when the current action started (elapsed time is client-side). */
   since: number;
-  /** Latest thumbnail, by attachment-store reference (never bytes on the wire). */
+  /** Latest thumbnail id (ephemeral, plugin-owned; never bytes on the wire). */
   thumbnailAttachmentId?: string;
   /** Summary of the most recent failed action (drives the red status dot). */
   lastError?: string;
@@ -52,28 +54,6 @@ export interface ObservationOptions {
   idleIntervalMs: number;
 }
 
-/** Structural view of the optional attachment seam (absent in headless compositions). */
-interface AttachmentLike {
-  saveImage(input: {
-    data: Uint8Array;
-    mediaType: string;
-    name?: string;
-  }): Promise<ImageAttachmentRefLike>;
-  readImage(
-    ref: ImageAttachmentRefLike,
-  ): Promise<{ data: Uint8Array; attachment: ImageAttachmentRefLike }>;
-}
-
-/** The attachment reference fields the store verifies reads against. */
-export interface ImageAttachmentRefLike {
-  attachmentId: unknown;
-  mediaType: string;
-  bytes: number;
-  width: number;
-  height: number;
-  name?: string;
-}
-
 /** Injectable clock/scheduler bits for tests. */
 export interface ObservationScheduler {
   setTimeout: (fn: () => void, ms: number) => unknown;
@@ -94,6 +74,8 @@ const DEFAULT_SCHEDULER: ObservationScheduler = {
 
 /** Max consecutive capture failures before a session drops to the idle cadence. */
 const FAILURE_BACKOFF_THRESHOLD = 3;
+/** Current + previous frame, so an in-flight HTTP fetch of the old id still lands. */
+const FRAME_RING_SIZE = 2;
 
 export class ObservationService {
   private readonly observations = new Map<string, SessionObservation>();
@@ -102,8 +84,14 @@ export class ObservationService {
   private readonly captureInFlight = new Set<string>();
   private readonly captureFailures = new Map<string, number>();
   private readonly lastActivity = new Map<string, number>();
-  /** attachmentId → full reference (the store requires it for verified reads). */
-  private readonly thumbRefs = new Map<string, ImageAttachmentRefLike>();
+  /** Ephemeral frame id → PNG bytes. Only the live ring members are present. */
+  private readonly frames = new Map<string, { data: Uint8Array; mediaType: string }>();
+  /** sessionId → oldest-first frame ids, length ≤ FRAME_RING_SIZE. */
+  private readonly rings = new Map<string, string[]>();
+  /** sessionId → sha256 of the current frame (skip republish when unchanged). */
+  private readonly hashes = new Map<string, string>();
+  /** sessionId → last issued sequence number. */
+  private readonly seqs = new Map<string, number>();
   /** Consecutive capture failures across ALL sessions (daemon-level signal). */
   private globalFailures = 0;
   private available = true;
@@ -179,10 +167,7 @@ export class ObservationService {
     this.cancelCapture(sessionId);
     this.captureFailures.delete(sessionId);
     this.lastActivity.delete(sessionId);
-    const entry = this.observations.get(sessionId);
-    if (entry?.thumbnailAttachmentId !== undefined) {
-      this.thumbRefs.delete(entry.thumbnailAttachmentId);
-    }
+    this.dropSessionFrames(sessionId);
     if (this.observations.delete(sessionId)) {
       this.emit({
         type: "remove",
@@ -243,24 +228,17 @@ export class ObservationService {
   }
 
   /**
-   * Read one captured thumbnail back through the store's verified path.
-   * Powers the plugin's own HTTP thumbnail route — frames are plugin-owned
-   * runtime data, never referenced by any session log, so the
-   * session-authorized client RPC cannot serve them.
+   * Read one captured thumbnail from the in-process ring. Powers the plugin's
+   * own HTTP thumbnail route — frames are plugin-owned runtime data, never
+   * referenced by any session log, so the session-authorized client RPC
+   * cannot serve them.
    */
   async readThumbnail(
     attachmentId: string,
   ): Promise<{ data: Uint8Array; mediaType: string } | undefined> {
-    const ref = this.thumbRefs.get(attachmentId);
-    if (ref === undefined) return undefined;
-    const attachments = this.deps.ctx.get("attachments") as AttachmentLike | undefined;
-    if (attachments === undefined) return undefined;
-    try {
-      const stored = await attachments.readImage(ref);
-      return { data: stored.data, mediaType: ref.mediaType };
-    } catch {
-      return undefined;
-    }
+    const frame = this.frames.get(attachmentId);
+    if (frame === undefined) return undefined;
+    return { data: frame.data, mediaType: frame.mediaType };
   }
 
   /** Tear down all state and timers (plugin dispose). */
@@ -271,7 +249,10 @@ export class ObservationService {
     this.captureInFlight.clear();
     this.captureFailures.clear();
     this.lastActivity.clear();
-    this.thumbRefs.clear();
+    this.frames.clear();
+    this.rings.clear();
+    this.hashes.clear();
+    this.seqs.clear();
     this.observations.clear();
     this.emit({ type: "reset" });
     this.listeners.clear();
@@ -313,8 +294,8 @@ export class ObservationService {
 
   /**
    * Capture one frame: `bsk screenshot --json` through the runner, bytes into
-   * the attachment store, reference onto the observation. Runs OUTSIDE the
-   * tool instrumentation on purpose — no action events, no registry writes.
+   * the in-process ring, id onto the observation. Runs OUTSIDE the tool
+   * instrumentation on purpose — no action events, no registry writes.
    * Failures keep the previous frame and back off silently.
    */
   private async capture(sessionId: string): Promise<void> {
@@ -322,11 +303,6 @@ export class ObservationService {
     const current = this.observations.get(sessionId);
     if (current === undefined || current.dead === true) return;
     if (this.captureInFlight.has(sessionId)) return;
-    const attachments = this.deps.ctx.get("attachments") as AttachmentLike | undefined;
-    if (attachments === undefined) {
-      // No attachment store (headless composition): thumbnails stay absent.
-      return;
-    }
     this.captureInFlight.add(sessionId);
     // One fixed scratch path per session (overwritten each frame) and always
     // unlinked after the bytes are read — /tmp must not grow without bound.
@@ -355,23 +331,12 @@ export class ObservationService {
       const reply = JSON.parse(result.stdout) as { path?: string };
       writtenPath = reply.path ?? outPath;
       const data = await readFile(writtenPath);
-      const ref = await attachments.saveImage({
-        data,
-        mediaType: "image/png",
-        name: `observation-${sessionId}.png`,
-      });
       const entry = this.observations.get(sessionId);
       if (entry === undefined) return;
       this.captureFailures.delete(sessionId);
       this.globalFailures = 0;
       this.setAvailable(true);
-      // Drop the replaced frame's reference — the store keeps the bytes, but
-      // this map (and the old id's read path) must not grow without bound.
-      if (entry.thumbnailAttachmentId !== undefined) {
-        this.thumbRefs.delete(entry.thumbnailAttachmentId);
-      }
-      this.thumbRefs.set(String(ref.attachmentId), ref);
-      this.put({ ...entry, thumbnailAttachmentId: String(ref.attachmentId) });
+      this.publishFrame(sessionId, entry, data);
     } catch {
       // Silent by design: keep the previous frame, count toward backoff.
       this.captureFailures.set(sessionId, (this.captureFailures.get(sessionId) ?? 0) + 1);
@@ -383,6 +348,33 @@ export class ObservationService {
       if (!this.disposed && this.observations.has(sessionId)) this.scheduleCapture(sessionId);
     }
   }
+
+  /** Insert a new frame, or no-op when the PNG bytes match the current one. */
+  private publishFrame(sessionId: string, entry: SessionObservation, data: Uint8Array): void {
+    const hash = createHash("sha256").update(data).digest("hex");
+    if (this.hashes.get(sessionId) === hash) return;
+    const seq = (this.seqs.get(sessionId) ?? 0) + 1;
+    this.seqs.set(sessionId, seq);
+    const id = `obs-${sessionId}-${seq}`;
+    this.frames.set(id, { data, mediaType: "image/png" });
+    this.hashes.set(sessionId, hash);
+    const ring = this.rings.get(sessionId) ?? [];
+    ring.push(id);
+    while (ring.length > FRAME_RING_SIZE) {
+      const evicted = ring.shift();
+      if (evicted !== undefined) this.frames.delete(evicted);
+    }
+    this.rings.set(sessionId, ring);
+    this.put({ ...entry, thumbnailAttachmentId: id });
+  }
+
+  private dropSessionFrames(sessionId: string): void {
+    for (const id of this.rings.get(sessionId) ?? []) this.frames.delete(id);
+    this.rings.delete(sessionId);
+    this.hashes.delete(sessionId);
+    this.seqs.delete(sessionId);
+  }
+
   /** Mark a session dead: grey it out, stop asking for frames, keep the entry for the strip. */
   private markDead(sessionId: string): void {
     const entry = this.observations.get(sessionId);

packages/dsh-plugin-browserskill/tests/observation.test.ts

@@ -58,7 +58,12 @@ interface FakeRunner extends BskRunner {
 
 /** Runner that answers screenshot with a real temp PNG and records kills. */
 function fakeRunner(
-  opts: { screenshotFails?: boolean; screenshotNotFound?: boolean; killCount?: number } = {},
+  opts: {
+    screenshotFails?: boolean;
+    screenshotNotFound?: boolean;
+    killCount?: number;
+    screenshotBytes?: () => Uint8Array;
+  } = {},
 ): FakeRunner {
   const calls: FakeRunner["calls"] = [];
   const killed: string[] = [];
@@ -82,7 +87,7 @@ function fakeRunner(
         }
         const outIndex = args.indexOf("--out") + 1;
         const out = args[outIndex];
-        writeFileSync(out, Buffer.from([0x89, 0x50, 0x4e, 0x47]));
+        writeFileSync(out, opts.screenshotBytes?.() ?? Buffer.from([0x89, 0x50, 0x4e, 0x47]));
         return {
           code: 0,
           stdout: JSON.stringify({
@@ -108,38 +113,20 @@ function fakeRunner(
   };
 }
 
-function fakeCtx(attachments?: unknown) {
-  return { get: (key: string) => (key === "attachments" ? attachments : undefined) } as never;
+function fakeCtx() {
+  return { get: () => undefined } as never;
 }
 
 function setup(opts: {
   runner?: FakeRunner;
-  attachments?: unknown;
   registry?: SessionRegistry;
   scheduler?: ReturnType<typeof fakeScheduler>;
 }) {
   const registry = opts.registry ?? new SessionRegistry(5);
   const runner = opts.runner ?? fakeRunner();
   const scheduler = opts.scheduler ?? fakeScheduler();
-  const attachments =
-    "attachments" in opts
-      ? opts.attachments
-      : {
-          saveImage: async (input: { data: Uint8Array; name?: string }) => ({
-            attachmentId: "att-1",
-            mediaType: "image/png",
-            bytes: input.data.byteLength,
-            width: 4,
-            height: 4,
-            name: input.name,
-          }),
-          readImage: async (ref: { attachmentId: unknown }) => ({
-            data: new Uint8Array([1, 2, 3]),
-            attachment: ref,
-          }),
-        };
   const service = new ObservationService({
-    ctx: fakeCtx(attachments),
+    ctx: fakeCtx(),
     runner,
     registry,
     queue: new KeyedExecutor(),
@@ -157,11 +144,6 @@ function own(registry: SessionRegistry, sessionId: string): void {
   registry.completeStart({ sessionId, startedAtMs: 1 });
 }
 
-async function flushMicrotasks(): Promise<void> {
-  // setImmediate yields to the poll phase so real fs reads inside capture() settle.
-  for (let i = 0; i < 20; i++) await new Promise((resolve) => setImmediate(resolve));
-}
-
 /** Poll until the condition holds (capture completion is real async I/O). */
 async function waitFor(cond: () => boolean, timeoutMs = 2000): Promise<void> {
   const start = Date.now();
@@ -218,7 +200,7 @@ describe("thumbnail cadence", () => {
     expect(scheduler.pending()).toEqual([0]);
     scheduler.runNext();
     await waitFor(() => captures() === 1 && scheduler.pending().length === 1);
-    expect(service.getState()[0].thumbnailAttachmentId).toBe("att-1");
+    expect(service.getState()[0].thumbnailAttachmentId).toBe("obs-s1-1");
     // Just captured with fresh activity: next frame on the fast cadence.
     expect(scheduler.pending()).toEqual([1500]);
     scheduler.runNext();
@@ -259,14 +241,14 @@ describe("thumbnail cadence", () => {
     expect(service.getState()[0].thumbnailAttachmentId).toBeUndefined();
   });
 
-  it("stays silent without an attachment store (headless composition)", async () => {
+  it("captures frames without an attachment store", async () => {
     const scheduler = fakeScheduler();
     const runner = fakeRunner();
-    const { service } = setup({ runner, scheduler, attachments: undefined });
+    const { service } = setup({ runner, scheduler });
     service.addSession("s1");
     scheduler.runNext();
-    await flushMicrotasks();
-    expect(runner.calls.filter((c) => c.args[0] === "screenshot")).toHaveLength(0);
+    await waitFor(() => service.getState()[0]?.thumbnailAttachmentId === "obs-s1-1");
+    expect((await service.readThumbnail("obs-s1-1"))?.mediaType).toBe("image/png");
   });
 });
 
@@ -286,7 +268,7 @@ describe("observation traffic isolation", () => {
     const captureEvent = captureEvents[0];
     if (captureEvent.type !== "upsert") throw new Error("unreachable");
     expect(captureEvent.session?.action).toBe("idle");
-    expect(captureEvent.session?.thumbnailAttachmentId).toBe("att-1");
+    expect(captureEvent.session?.thumbnailAttachmentId).toBe("obs-s1-1");
     // …and the registry's current pointer never moved through observation.
     expect(registry.current()).toBe("s1");
     expect(registry.list()[0].startedAtMs).toBe(1);
@@ -551,9 +533,9 @@ describe("thumbnail read-back", () => {
     service.addSession("s1");
     scheduler.runNext();
     await waitFor(() => service.getState()[0]?.thumbnailAttachmentId !== undefined);
-    const frame = await service.readThumbnail("att-1");
+    const frame = await service.readThumbnail("obs-s1-1");
     expect(frame?.mediaType).toBe("image/png");
-    expect([...(frame?.data ?? [])]).toEqual([1, 2, 3]);
+    expect([...(frame?.data ?? [])]).toEqual([0x89, 0x50, 0x4e, 0x47]);
     expect(await service.readThumbnail("nope")).toBeUndefined();
     service.dispose();
   });
@@ -598,47 +580,65 @@ describe("scratch files and thumbnail references", () => {
     service.dispose();
   });
 
-  it("drops the replaced frame reference and clears it on remove", async () => {
-    let counter = 0;
-    const attachments = {
-      saveImage: async (input: { data: Uint8Array; name?: string }) => {
-        counter += 1;
-        return {
-          attachmentId: `att-${counter}`,
-          mediaType: "image/png",
-          bytes: input.data.byteLength,
-          width: 4,
-          height: 4,
-          name: input.name,
-        };
-      },
-      readImage: async (ref: { attachmentId: unknown }) => ({
-        data: new Uint8Array([1, 2, 3]),
-        attachment: ref,
-      }),
-    };
+  it("keeps the last two distinct frames and evicts the third", async () => {
+    let n = 0;
+    const runner = fakeRunner({
+      screenshotBytes: () => Uint8Array.from([0x89, 0x50, 0x4e, 0x47, n++]),
+    });
     const scheduler = fakeScheduler();
-    const { service } = setup({ scheduler, attachments });
+    const { service } = setup({ scheduler, runner });
     service.addSession("s1");
     scheduler.runNext();
     await waitFor(
       () =>
-        service.getState()[0]?.thumbnailAttachmentId === "att-1" &&
+        service.getState()[0]?.thumbnailAttachmentId === "obs-s1-1" &&
         scheduler.pending().length === 1,
     );
     service.endAction("s1");
     scheduler.runNext();
     await waitFor(
       () =>
-        service.getState()[0]?.thumbnailAttachmentId === "att-2" &&
+        service.getState()[0]?.thumbnailAttachmentId === "obs-s1-2" &&
         scheduler.pending().length === 1,
     );
-    // the replaced frame's read path is gone
-    expect(await service.readThumbnail("att-1")).toBeUndefined();
-    expect((await service.readThumbnail("att-2"))?.mediaType).toBe("image/png");
-    // removing the session clears its live reference too
+    expect((await service.readThumbnail("obs-s1-1"))?.mediaType).toBe("image/png");
+    expect((await service.readThumbnail("obs-s1-2"))?.mediaType).toBe("image/png");
+    service.endAction("s1");
+    scheduler.runNext();
+    await waitFor(
+      () =>
+        service.getState()[0]?.thumbnailAttachmentId === "obs-s1-3" &&
+        scheduler.pending().length === 1,
+    );
+    expect(await service.readThumbnail("obs-s1-1")).toBeUndefined();
+    expect((await service.readThumbnail("obs-s1-2"))?.mediaType).toBe("image/png");
+    expect((await service.readThumbnail("obs-s1-3"))?.mediaType).toBe("image/png");
     service.removeSession("s1");
-    expect(await service.readThumbnail("att-2")).toBeUndefined();
+    expect(await service.readThumbnail("obs-s1-2")).toBeUndefined();
+    expect(await service.readThumbnail("obs-s1-3")).toBeUndefined();
+    service.dispose();
+  });
+
+  it("reuses the current frame id when the PNG bytes are unchanged", async () => {
+    const scheduler = fakeScheduler();
+    const runner = fakeRunner();
+    const { service } = setup({ scheduler, runner });
+    service.addSession("s1");
+    scheduler.runNext();
+    await waitFor(
+      () =>
+        service.getState()[0]?.thumbnailAttachmentId === "obs-s1-1" &&
+        scheduler.pending().length === 1,
+    );
+    service.endAction("s1");
+    scheduler.runNext();
+    await waitFor(
+      () =>
+        runner.calls.filter((c) => c.args[0] === "screenshot").length >= 2 &&
+        scheduler.pending().length === 1,
+    );
+    expect(service.getState()[0]?.thumbnailAttachmentId).toBe("obs-s1-1");
+    expect((await service.readThumbnail("obs-s1-1"))?.mediaType).toBe("image/png");
     service.dispose();
   });
 });

packages/dsh-plugin-browserskill/tests/tools.test.ts

@@ -733,6 +733,10 @@ describe("observation action instrumentation timing", () => {
     const navigateGate = new Promise<void>((resolve) => {
       releaseNavigate = resolve;
     });
+    let releaseSnapshot!: () => void;
+    const snapshotGate = new Promise<void>((resolve) => {
+      releaseSnapshot = resolve;
+    });
     const runner: BskRunner = {
       async run(args: string[]): Promise<BskRunResult> {
         if (args[0] === "session") {
@@ -754,6 +758,21 @@ describe("observation action instrumentation timing", () => {
             aborted: false,
           };
         }
+        if (args[0] === "snapshot") {
+          await snapshotGate;
+          return {
+            code: 0,
+            stdout: JSON.stringify({ text: "ok", ref_count: 0, tab_id: 7, truncated: false }),
+            stderr: "",
+            timedOut: false,
+            aborted: false,
+          };
+        }
+        // Observation captures share the session queue; fail them fast so they
+        // cannot stall the action-label assertions.
+        if (args[0] === "screenshot") {
+          return { code: 1, stdout: "", stderr: "skip", timedOut: false, aborted: false };
+        }
         return {
           code: 0,
           stdout: JSON.stringify({ text: "ok", ref_count: 0, tab_id: 7, truncated: false }),
@@ -792,6 +811,7 @@ describe("observation action instrumentation timing", () => {
     await vi.waitFor(() =>
       expect(observation.getState().find((s) => s.sessionId === "s1")?.action).toBe("snapshotting"),
     );
+    releaseSnapshot();
     await b;
     await vi.waitFor(() =>
       expect(observation.getState().find((s) => s.sessionId === "s1")?.action).toBe("idle"),