diff --git a/packages/dsh-plugin-browserskill/src/observation.ts b/packages/dsh-plugin-browserskill/src/observation.ts index afc761a..78991b6 100644 --- a/packages/dsh-plugin-browserskill/src/observation.ts +++ b/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; - 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(); @@ -102,8 +84,14 @@ export class ObservationService { private readonly captureInFlight = new Set(); private readonly captureFailures = new Map(); private readonly lastActivity = new Map(); - /** attachmentId → full reference (the store requires it for verified reads). */ - private readonly thumbRefs = new Map(); + /** Ephemeral frame id → PNG bytes. Only the live ring members are present. */ + private readonly frames = new Map(); + /** sessionId → oldest-first frame ids, length ≤ FRAME_RING_SIZE. */ + private readonly rings = new Map(); + /** sessionId → sha256 of the current frame (skip republish when unchanged). */ + private readonly hashes = new Map(); + /** sessionId → last issued sequence number. */ + private readonly seqs = new Map(); /** 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 { @@ -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); diff --git a/packages/dsh-plugin-browserskill/tests/observation.test.ts b/packages/dsh-plugin-browserskill/tests/observation.test.ts index ce6f79b..e67f768 100644 --- a/packages/dsh-plugin-browserskill/tests/observation.test.ts +++ b/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; }) { 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 { - // 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 { 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(); }); }); diff --git a/packages/dsh-plugin-browserskill/tests/tools.test.ts b/packages/dsh-plugin-browserskill/tests/tools.test.ts index 5762748..9ecb14b 100644 --- a/packages/dsh-plugin-browserskill/tests/tools.test.ts +++ b/packages/dsh-plugin-browserskill/tests/tools.test.ts @@ -733,6 +733,10 @@ describe("observation action instrumentation timing", () => { const navigateGate = new Promise((resolve) => { releaseNavigate = resolve; }); + let releaseSnapshot!: () => void; + const snapshotGate = new Promise((resolve) => { + releaseSnapshot = resolve; + }); const runner: BskRunner = { async run(args: string[]): Promise { 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"),