diff --git a/docs/plugins.md b/docs/plugins.md index 951738c902..f07f831803 100644 --- a/docs/plugins.md +++ b/docs/plugins.md @@ -382,6 +382,15 @@ grace. During daemon startup, plugin sessions may connect while application WebS paused; the daemon accepts clients only after configured plugins have settled and the initial catalog is complete. +The app retains client bundles per host in its file cache (IndexedDB on web), keyed by plugin ID +and the daemon's SHA-256. A cold client verifies stored contents before advertising their hashes; +missing or corrupt files request full bundles. Only the latest catalog's bundles remain. Hashing +uses Expo's native async digest on mobile, avoiding a JavaScript hash loop on Hermes; reading and +passing the bundle string still costs memory and time once per client lifetime. +Persistence runs after catalog reconstruction without delaying installation. One write runs at a +time; while storage is busy, only the newest pending catalog is kept. A failed write leaves the +in-memory bundles usable, and a later catalog retries persistence. + When the same plugin contribution exists on multiple hosts, Paseo shows it once in the sidebar and adds a host picker to the screen header. The selected host supplies the bundle, RPC transport, and query cache. Explicit SDK targets follow the [host API contract](../public-docs/plugins/reference.md#discover-hosts-and-target-another-host). diff --git a/packages/app/e2e/browser/plugin-timeline.spec.ts b/packages/app/e2e/browser/plugin-timeline.spec.ts index 166e6b9881..d75653298e 100644 --- a/packages/app/e2e/browser/plugin-timeline.spec.ts +++ b/packages/app/e2e/browser/plugin-timeline.spec.ts @@ -1,4 +1,5 @@ import { test } from "../support/fixtures"; +import { installDaemonWebSocketGate } from "../support/helpers/daemon-websocket-gate"; import { withTimelinePlugin, viewInlineThinkingExample, @@ -6,6 +7,8 @@ import { interactWithStreamingCard, expectWholeCompletedCard, expectBothConsecutiveTools, + trackPluginChatRecovery, + reconnectPluginChat, } from "../support/helpers/plugin-timeline"; for (const width of [1100, 390]) { @@ -13,13 +16,24 @@ for (const width of [1100, 390]) { page, }, info) => { await page.setViewportSize({ width, height: 900 }); + const gate = await installDaemonWebSocketGate(page); await withTimelinePlugin(page, info, "assistant", async (agent) => { await requestPluginTimeline(agent); await interactWithStreamingCard(page); await agent.client.waitForFinish(agent.agentId, 30_000); await expectWholeCompletedCard(page); // Local state is checked during growth; the viewport remounts on the history handoff. + const recovery = trackPluginChatRecovery(page); + gate.holdTimelineResponses(agent.agentId); await page.reload({ waitUntil: "domcontentloaded" }); + await gate.waitForHeldTimelineResponse(); + await recovery.expectCatalogWithoutRetransmission(); + gate.releaseHeldTimelineResponses(); + await expectWholeCompletedCard(page); + await recovery.expectRecoveredWithoutRetransmission(); + recovery.reset(); + await reconnectPluginChat(page, gate); + await recovery.expectRecoveredWithoutRetransmission(); await expectWholeCompletedCard(page); }); }); diff --git a/packages/app/e2e/support/helpers/plugin-timeline.ts b/packages/app/e2e/support/helpers/plugin-timeline.ts index 533b546022..98ab7d303b 100644 --- a/packages/app/e2e/support/helpers/plugin-timeline.ts +++ b/packages/app/e2e/support/helpers/plugin-timeline.ts @@ -5,8 +5,64 @@ import { expect, type Page, type TestInfo } from "@playwright/test"; import { copyPluginExample, pluginRequirements } from "./plugin-fixture"; import { connectNewWorkspaceDaemonClient } from "./new-workspace"; import { openAgentRoute, seedMockAgentWorkspace, type MockAgentWorkspace } from "./mock-agent"; +import { installDaemonWebSocketGate } from "./daemon-websocket-gate"; +import { expectReconnectingToastVisible } from "./workspace-ui"; const PLUGIN_ID = "timeline-transform-e2e"; + +/** Observe the app's real socket, including fresh sockets after a reload. */ +export function trackPluginChatRecovery(page: Page) { + const messages: { + type: string; + payload?: { plugins?: { id: string; clientBundle?: string }[] }; + }[] = []; + page.on("websocket", (socket) => { + const collect = ({ payload }: { payload: string | Buffer }) => { + if (typeof payload !== "string") return; + const frame = JSON.parse(payload); + if (frame.type === "session") messages.push(frame.message); + }; + socket.on("framesent", collect); + socket.on("framereceived", collect); + }); + return { + reset() { + messages.length = 0; + }, + async expectCatalogWithoutRetransmission() { + await expect + .poll(() => + messages.some( + (message) => + message.type === "plugin.catalog.get.response" && + message.payload?.plugins?.some((plugin) => plugin.id === PLUGIN_ID), + ), + ) + .toBe(true); + const entry = messages + .find((message) => message.type === "plugin.catalog.get.response") + ?.payload?.plugins?.find((plugin) => plugin.id === PLUGIN_ID); + expect(entry).toBeDefined(); + expect(entry).not.toHaveProperty("clientBundle"); + }, + async expectRecoveredWithoutRetransmission() { + await this.expectCatalogWithoutRetransmission(); + await expect + .poll(() => messages.some((message) => message.type === "fetch_agent_timeline_response")) + .toBe(true); + }, + }; +} + +export async function reconnectPluginChat( + page: Page, + gate: Awaited>, +) { + await gate.drop(); + await gate.waitForBlockedConnection(); + await expectReconnectingToastVisible(page); + gate.restore(); +} export const ASSISTANT_TEXT = "Skill header\n\n# Instructions\n\nRead every paragraph and preserve the full message while this response is still streaming."; diff --git a/packages/app/src/plugins/client-bundles/index.ts b/packages/app/src/plugins/client-bundles/index.ts new file mode 100644 index 0000000000..06580dd222 --- /dev/null +++ b/packages/app/src/plugins/client-bundles/index.ts @@ -0,0 +1,32 @@ +import * as FileSystem from "expo-file-system/legacy"; +import { CryptoDigestAlgorithm, digestStringAsync, randomUUID } from "expo-crypto"; +import { createPluginClientBundleStore } from "./store"; + +export function createHostPluginClientBundleStore(serverId: string) { + const directory = `${FileSystem.cacheDirectory}plugin-client-bundles/${encodeURIComponent(serverId)}/`; + const ensureDirectory = async () => { + if (!FileSystem.cacheDirectory) throw new Error("Plugin bundle cache directory unavailable"); + await FileSystem.makeDirectoryAsync(directory, { intermediates: true }); + }; + return createPluginClientBundleStore( + { + async list() { + await ensureDirectory(); + return FileSystem.readDirectoryAsync(directory); + }, + read: (name) => FileSystem.readAsStringAsync(`${directory}${name}`), + async writeAtomic(name, contents) { + await ensureDirectory(); + const temporary = `${directory}${randomUUID()}.tmp`; + try { + await FileSystem.writeAsStringAsync(temporary, contents); + await FileSystem.moveAsync({ from: temporary, to: `${directory}${name}` }); + } finally { + await FileSystem.deleteAsync(temporary, { idempotent: true }); + } + }, + remove: (name) => FileSystem.deleteAsync(`${directory}${name}`, { idempotent: true }), + }, + (contents) => digestStringAsync(CryptoDigestAlgorithm.SHA256, contents), + ); +} diff --git a/packages/app/src/plugins/client-bundles/index.web.ts b/packages/app/src/plugins/client-bundles/index.web.ts new file mode 100644 index 0000000000..6e67778089 --- /dev/null +++ b/packages/app/src/plugins/client-bundles/index.web.ts @@ -0,0 +1,62 @@ +import { createPluginClientBundleStore } from "./store"; + +const DATABASE_NAME = "paseo-plugin-client-bundles"; +const STORE_NAME = "bundles"; +let opening: Promise | undefined; + +function openDatabase(): Promise { + opening ??= new Promise((resolve, reject) => { + const request = indexedDB.open(DATABASE_NAME, 1); + request.addEventListener("upgradeneeded", () => request.result.createObjectStore(STORE_NAME)); + request.addEventListener("success", () => resolve(request.result)); + request.addEventListener("error", () => reject(request.error)); + }).catch((error) => { + opening = undefined; + throw error; + }); + return opening; +} + +export function createHostPluginClientBundleStore(serverId: string) { + const prefix = `${encodeURIComponent(serverId)}/`; + const range = () => IDBKeyRange.bound(prefix, `${prefix}\uffff`); + async function run( + mode: IDBTransactionMode, + operation: (store: IDBObjectStore) => IDBRequest, + ): Promise { + const database = await openDatabase(); + return new Promise((resolve, reject) => { + const transaction = database.transaction(STORE_NAME, mode); + const request = operation(transaction.objectStore(STORE_NAME)); + transaction.addEventListener("complete", () => resolve(request.result)); + transaction.addEventListener("abort", () => reject(transaction.error)); + transaction.addEventListener("error", () => reject(transaction.error)); + }); + } + return createPluginClientBundleStore( + { + async list() { + return (await run("readonly", (store) => store.getAllKeys(range()))) + .filter((key): key is string => typeof key === "string") + .map((key) => key.slice(prefix.length)); + }, + async read(name) { + const result: unknown = await run("readonly", (store) => store.get(`${prefix}${name}`)); + if (typeof result !== "string") throw new Error("Missing plugin client bundle"); + return result; + }, + async writeAtomic(name, contents) { + await run("readwrite", (store) => store.put(contents, `${prefix}${name}`)); + }, + async remove(name) { + await run("readwrite", (store) => store.delete(`${prefix}${name}`)); + }, + }, + async (contents) => { + const digest = await crypto.subtle.digest("SHA-256", new TextEncoder().encode(contents)); + return Array.from(new Uint8Array(digest), (byte) => byte.toString(16).padStart(2, "0")).join( + "", + ); + }, + ); +} diff --git a/packages/app/src/plugins/client-bundles/storage.browser.test.ts b/packages/app/src/plugins/client-bundles/storage.browser.test.ts new file mode 100644 index 0000000000..9acd7e0c52 --- /dev/null +++ b/packages/app/src/plugins/client-bundles/storage.browser.test.ts @@ -0,0 +1,36 @@ +import { expect, test } from "vitest"; +import { createHostPluginClientBundleStore } from "./index.web"; + +async function sha256(contents: string) { + const digest = await crypto.subtle.digest("SHA-256", new TextEncoder().encode(contents)); + return Array.from(new Uint8Array(digest), (byte) => byte.toString(16).padStart(2, "0")).join(""); +} + +test("browser bundles survive a fresh store and remain isolated by host", async () => { + const first = createHostPluginClientBundleStore("bundle-test-host-a"); + const second = createHostPluginClientBundleStore("bundle-test-host-b"); + const clientBundle = "bundle".repeat(250_000); + const entry = { + id: "same-plugin", + clientBundle, + clientBundleHash: await sha256(clientBundle), + }; + const otherCode = "other host code"; + const other = { + id: entry.id, + clientBundle: otherCode, + clientBundleHash: await sha256(otherCode), + }; + try { + await first.replace([entry]); + await second.replace([other]); + expect(await createHostPluginClientBundleStore("bundle-test-host-a").load()).toEqual([entry]); + expect(await createHostPluginClientBundleStore("bundle-test-host-b").load()).toEqual([other]); + await first.replace([]); + expect(await createHostPluginClientBundleStore("bundle-test-host-a").load()).toEqual([]); + expect(await second.load()).toEqual([other]); + } finally { + await first.replace([]); + await second.replace([]); + } +}); diff --git a/packages/app/src/plugins/client-bundles/store.test.ts b/packages/app/src/plugins/client-bundles/store.test.ts new file mode 100644 index 0000000000..c3332c6b0e --- /dev/null +++ b/packages/app/src/plugins/client-bundles/store.test.ts @@ -0,0 +1,179 @@ +import { createHash } from "node:crypto"; +import { DaemonClient } from "@getpaseo/client/internal/daemon-client"; +import { expect, test } from "vitest"; +import { createPluginClientBundleStore, type PluginBundleFiles } from "./store"; + +const sha256 = async (contents: string) => createHash("sha256").update(contents).digest("hex"); + +class BundleFiles implements PluginBundleFiles { + readonly contents = new Map(); + readonly writes: string[] = []; + async list() { + return [...this.contents.keys()]; + } + async read(name: string) { + const contents = this.contents.get(name); + if (contents === undefined) throw new Error("File disappeared"); + return contents; + } + async writeAtomic(name: string, contents: string) { + this.writes.push(name); + this.contents.set(name, contents); + } + async remove(name: string) { + this.contents.delete(name); + } +} + +async function bundle(id: string, clientBundle: string) { + return { id, clientBundle, clientBundleHash: await sha256(clientBundle) }; +} + +test("a cold store loads verified bundles and prunes versions missing from the latest catalog", async () => { + const files = new BundleFiles(); + const old = await bundle("changed", "old"); + const kept = await bundle("kept", "Unicode: 折叠屏 🐚"); + await createPluginClientBundleStore(files, sha256).replace([old, kept]); + const cold = createPluginClientBundleStore(files, sha256); + expect(await cold.load()).toEqual([old, kept]); + const changed = await bundle("changed", "new"); + await cold.replace([kept, changed]); + expect(files.contents.size).toBe(2); + expect(await createPluginClientBundleStore(files, sha256).load()).toEqual([kept, changed]); + expect(files.writes).toHaveLength(3); + await cold.replace([]); + expect(files.contents.size).toBe(0); +}); + +test("same-length corruption, truncated files and unreadable files are not advertised", async () => { + const files = new BundleFiles(); + const entries = await Promise.all([ + bundle("same-length", "good"), + bundle("truncated", "long"), + bundle("missing", "lost"), + ]); + await createPluginClientBundleStore(files, sha256).replace(entries); + const names = [...files.contents.keys()]; + files.contents.set(names[0]!, "evil"); + files.contents.set(names[1]!, "lo"); + // The directory entry still exists while reading the file fails. + const unreadable: PluginBundleFiles = { + list: () => files.list(), + read: (name) => + name === names[2] ? Promise.reject(new Error("Unreadable")) : files.read(name), + writeAtomic: (name, contents) => files.writeAtomic(name, contents), + remove: (name) => files.remove(name), + }; + const cold = createPluginClientBundleStore(unreadable, sha256); + expect(await cold.load()).toEqual([]); + await cold.replace(entries); + expect(await createPluginClientBundleStore(files, sha256).load()).toEqual(entries); +}); + +test("an interrupted atomic write leaves the prior catalog available for a cold reader", async () => { + const files = new BundleFiles(); + const previous = await bundle("plugin", "previous"); + await createPluginClientBundleStore(files, sha256).replace([previous]); + const failing: PluginBundleFiles = { + list: () => files.list(), + read: (name) => files.read(name), + remove: (name) => files.remove(name), + writeAtomic: async () => { + throw new Error("Disk full"); + }, + }; + await expect( + createPluginClientBundleStore(failing, sha256).replace([await bundle("plugin", "next")]), + ).rejects.toThrow("Disk full"); + expect(await createPluginClientBundleStore(files, sha256).load()).toEqual([previous]); +}); + +// Exercise the app store through a fresh platform-agnostic client, using its transport port. +test.each(["valid", "corrupt", "missing"])( + "cold catalog uses a %s persisted bundle correctly", + async (state) => { + const files = new BundleFiles(); + const entry = await bundle("plugin", "good code"); + await createPluginClientBundleStore(files, sha256).replace([entry]); + if (state === "corrupt") files.contents.set([...files.contents.keys()][0]!, "evil code"); + if (state === "missing") files.contents.clear(); + let open = () => {}; + let receive: (data: unknown, binary: boolean) => void = () => {}; + const requests: Record[] = []; + const client = new DaemonClient({ + url: "ws://test", + clientId: "plugin-cache-test", + reconnect: { enabled: false }, + createPluginClientBundleStore: () => createPluginClientBundleStore(files, sha256), + transportFactory: () => ({ + send(data) { + if (typeof data !== "string") return; + const frame = JSON.parse(data); + if (frame.type !== "session") return; + const request = frame.message; + if (request.type !== "plugin.catalog.get.request") return; + requests.push(request); + const { clientBundle: _code, ...reference } = entry; + receive( + JSON.stringify({ + type: "session", + message: { + type: "plugin.catalog.get.response", + payload: { + requestId: request.requestId, + plugins: [request.cachedClientBundles ? reference : entry], + }, + }, + }), + false, + ); + }, + close() {}, + onOpen(handler) { + open = handler; + return () => {}; + }, + onMessage(handler) { + receive = handler; + return () => {}; + }, + onClose() { + return () => {}; + }, + onError() { + return () => {}; + }, + }), + }); + try { + const connecting = client.connect(); + open(); + receive( + JSON.stringify({ + type: "session", + message: { + type: "status", + payload: { + status: "server_info", + serverId: "cache-host", + hostname: null, + version: null, + features: { plugins: true }, + }, + }, + }), + false, + ); + await connecting; + expect(await client.getPluginCatalog()).toEqual([entry]); + if (state === "valid") + expect(requests[0]?.cachedClientBundles).toEqual([ + { id: entry.id, clientBundleHash: entry.clientBundleHash }, + ]); + else expect(requests[0]).not.toHaveProperty("cachedClientBundles"); + await expect.poll(() => createPluginClientBundleStore(files, sha256).load()).toEqual([entry]); + } finally { + await client.close(); + } + }, +); diff --git a/packages/app/src/plugins/client-bundles/store.ts b/packages/app/src/plugins/client-bundles/store.ts new file mode 100644 index 0000000000..4a120c2c2a --- /dev/null +++ b/packages/app/src/plugins/client-bundles/store.ts @@ -0,0 +1,56 @@ +import type { + PluginClientBundle, + PluginClientBundleStore, +} from "@getpaseo/client/internal/daemon-client"; + +/** Each adapter is scoped to one host. Writes publish a complete file/record atomically. */ +export interface PluginBundleFiles { + list(): Promise; + read(name: string): Promise; + writeAtomic(name: string, contents: string): Promise; + remove(name: string): Promise; +} + +export function createPluginClientBundleStore( + files: PluginBundleFiles, + sha256: (contents: string) => Promise, +): PluginClientBundleStore { + const verified = new Set(); + const filename = (bundle: PluginClientBundle) => + `${encodeURIComponent(bundle.id)}.${bundle.clientBundleHash}.js`; + return { + async load() { + const bundles: PluginClientBundle[] = []; + for (const name of await files.list()) { + try { + const match = /^(.*)\.([a-f0-9]{64})\.js$/.exec(name); + if (!match) continue; + const id = decodeURIComponent(match[1]!); + const clientBundleHash = match[2]!; + const clientBundle = await files.read(name); + if ((await sha256(clientBundle)) !== clientBundleHash) continue; + verified.add(name); + bundles.push({ id, clientBundleHash, clientBundle }); + } catch { + // Missing, unreadable or corrupt files are cache misses, never executable code. + } + } + return bundles; + }, + async replace(bundles) { + const retained = new Set(); + for (const bundle of bundles) { + const name = filename(bundle); + retained.add(name); + if (verified.has(name)) continue; + await files.writeAtomic(name, bundle.clientBundle); + verified.add(name); + } + for (const name of await files.list()) { + if (retained.has(name)) continue; + await files.remove(name); + verified.delete(name); + } + }, + }; +} diff --git a/packages/app/src/runtime/host-runtime.ts b/packages/app/src/runtime/host-runtime.ts index bb953220cb..0c7c9f4660 100644 --- a/packages/app/src/runtime/host-runtime.ts +++ b/packages/app/src/runtime/host-runtime.ts @@ -1,3 +1,4 @@ +import { createHostPluginClientBundleStore } from "@/plugins/client-bundles"; import { watchdog } from "@/watchdog"; import { createHostChannelTransport } from "@getpaseo/client/internal/hosts"; import type { OwnedSubscription } from "@getpaseo/client"; @@ -589,6 +590,7 @@ function createDefaultDeps( capabilities: appCapabilities, trace: createPerformanceTrace({ host: host.serverId, connection: connection.id }), providerSnapshots: "wire", + createPluginClientBundleStore: createHostPluginClientBundleStore, ...(host.password ? { password: host.password } : {}), localCredential: () => readDesktopManagedLocalCredential(connection), } satisfies Omit; @@ -668,6 +670,7 @@ function createDefaultDeps( ), reconnect: { enabled: false }, appVersion: resolveAppVersion() ?? undefined, + createPluginClientBundleStore: createHostPluginClientBundleStore, }); try { await client.connect(); @@ -684,6 +687,7 @@ function createDefaultDeps( ...(timeoutMs !== undefined ? { timeoutMs } : {}), capabilities: appCapabilities, trace: createPerformanceTrace({ host: host.serverId, connection: connection.id }), + createPluginClientBundleStore: createHostPluginClientBundleStore, }); }, getClientId: () => getOrCreateClientId(), diff --git a/packages/app/src/timeline/replica.test.ts b/packages/app/src/timeline/replica.test.ts index 47bd36596a..0d825673e9 100644 --- a/packages/app/src/timeline/replica.test.ts +++ b/packages/app/src/timeline/replica.test.ts @@ -371,6 +371,22 @@ describe("viewed timeline persistence", () => { ]); }); + it("preparing an already painted cache again keeps history out of the live head", async () => { + useSessionStore.getState().initializeSession(SERVER_ID, null); + const replica = createTimelineReplica({ + serverId: SERVER_ID, + storage: { readTimeline: async () => cachedTimeline(), commitTimeline: () => undefined }, + prepareAgent: async () => undefined, + }); + await replica.prepare(AGENT_ID); + const painted = useSessionStore.getState().sessions[SERVER_ID]!.agentStreamTail.get(AGENT_ID); + expect(painted).toEqual(cachedTimeline().items); + await replica.prepare(AGENT_ID); + const session = useSessionStore.getState().sessions[SERVER_ID]!; + expect(session.agentStreamHead.get(AGENT_ID) ?? []).toEqual([]); + expect(session.agentStreamTail.get(AGENT_ID)).toBe(painted); + }); + it("reconciles a live head that overlaps the cached canonical timeline", async () => { useSessionStore.getState().initializeSession(SERVER_ID, null); let release!: (value: CachedTimeline) => void; diff --git a/packages/app/src/timeline/viewed-timeline-sync.ts b/packages/app/src/timeline/viewed-timeline-sync.ts index 5939feabce..2e14a42093 100644 --- a/packages/app/src/timeline/viewed-timeline-sync.ts +++ b/packages/app/src/timeline/viewed-timeline-sync.ts @@ -109,6 +109,15 @@ class TimelineReplicaOwner implements TimelineReplica { async prepare(agentId: string): Promise { const existing = this.preparations.get(agentId); if (existing) return existing; + const timeline = selectAgentTimelineState( + useSessionStore.getState().sessions[this.serverId], + agentId, + ); + // A route and the visible-chat owner can both prepare before catch-up finishes. + // Repainting that same cache would treat its history as a live head. + if (timeline.status !== "cold" && this.cachedRanges.has(agentId)) { + return; + } const preparation = this.load(agentId).finally(() => { if (this.preparations.get(agentId) === preparation) { this.preparations.delete(agentId); diff --git a/packages/app/src/utils/test-daemon-connection.test.ts b/packages/app/src/utils/test-daemon-connection.test.ts index 4dc82dda29..f528065437 100644 --- a/packages/app/src/utils/test-daemon-connection.test.ts +++ b/packages/app/src/utils/test-daemon-connection.test.ts @@ -129,6 +129,23 @@ describe("test-daemon-connection connectToDaemon", () => { expect(probe.createdConfigs()[0]?.transportFactory).toBeUndefined(); }); + it("keeps the bundle cache factory on a probe client that the host runtime can adopt", async () => { + const { connectToDaemon } = await import("./test-daemon-connection"); + const createPluginClientBundleStore = () => ({ + async load() { + return []; + }, + async replace() {}, + }); + const { client } = await connectToDaemon( + { id: "direct:lan:6767", type: "directTcp", endpoint: "lan:6767" }, + { createPluginClientBundleStore }, + probe.deps, + ); + expect(client.config.createPluginClientBundleStore).toBe(createPluginClientBundleStore); + await client.close(); + }); + it("encodes the local socket target into the client config", async () => { const { connectToDaemon } = await import("./test-daemon-connection"); const result = await connectToDaemon( diff --git a/packages/app/src/utils/test-daemon-connection.ts b/packages/app/src/utils/test-daemon-connection.ts index 18ffa62dc4..d292c67001 100644 --- a/packages/app/src/utils/test-daemon-connection.ts +++ b/packages/app/src/utils/test-daemon-connection.ts @@ -132,12 +132,13 @@ function resolveConnectionCredentials( export async function buildClientConfig( connection: HostConnection, serverId?: string, - options?: { + options: { capabilities?: DaemonClientConfig["capabilities"]; trace?: DaemonClientConfig["trace"]; + createPluginClientBundleStore?: DaemonClientConfig["createPluginClientBundleStore"]; password?: string; localCredential?: DaemonClientConfig["localCredential"]; - }, + } = {}, deps: Pick< DaemonConnectionDependencies, | "getClientId" @@ -157,8 +158,9 @@ export async function buildClientConfig( suppressSendErrors: true, reconnect: { enabled: false }, ...resolveConnectionCredentials(connection, options), - ...(options?.capabilities ? { capabilities: options.capabilities } : {}), - ...(options?.trace ? { trace: options.trace } : {}), + ...(options.capabilities ? { capabilities: options.capabilities } : {}), + ...(options.trace ? { trace: options.trace } : {}), + createPluginClientBundleStore: options.createPluginClientBundleStore, ...((connection.type === "directSocket" || connection.type === "directPipe") && desktopTransportFactory ? { transportFactory: desktopTransportFactory } @@ -288,6 +290,7 @@ interface ProbeOptions { localCredential?: DaemonClientConfig["localCredential"]; capabilities?: DaemonClientConfig["capabilities"]; trace?: DaemonClientConfig["trace"]; + createPluginClientBundleStore?: DaemonClientConfig["createPluginClientBundleStore"]; } function resolveTimeout(connection: HostConnection, options?: ProbeOptions): number { diff --git a/packages/client/src/daemon-client.test.ts b/packages/client/src/daemon-client.test.ts index 2a3b409cee..112113caf6 100644 --- a/packages/client/src/daemon-client.test.ts +++ b/packages/client/src/daemon-client.test.ts @@ -1,5 +1,5 @@ import { readFileSync } from "node:fs"; -import { afterEach, expect, expectTypeOf, test, vi } from "vitest"; +import { afterEach, describe, expect, expectTypeOf, test, vi } from "vitest"; import { z } from "zod"; import { DaemonClient, @@ -9,6 +9,8 @@ import { type CreateAgentRequestOptions, type DaemonTransport, type Logger, + type PluginClientBundle, + type PluginClientBundleStore, } from "./daemon-client"; import { CLIENT_CAPS } from "@getpaseo/protocol/client-capabilities"; import { BROWSER_AUTOMATION_COMMAND_NAMES } from "@getpaseo/protocol/browser-automation/rpc-schemas"; @@ -2099,6 +2101,188 @@ test("gets a structured plugin log snapshot", async () => { ]); }); +describe("plugin catalog", () => { + async function connectedClient(createStore?: (serverId: string) => PluginClientBundleStore) { + const mock = createMockTransport(); + const client = new DaemonClient({ + url: "ws://test", + clientId: "clsk_unit_test", + logger: createMockLogger(), + createPluginClientBundleStore: createStore, + reconnect: { enabled: false }, + transportFactory: () => mock.transport, + }); + clients.push(client); + const connectPromise = client.connect(); + mock.triggerOpen({ features: { plugins: true } }); + await connectPromise; + return { client, mock }; + } + + async function fetchCatalog( + harness: Awaited>, + respond: (request: Record) => Array>, + ) { + const before = harness.mock.sent.length; + const response = harness.client.getPluginCatalog(); + await vi.waitFor(() => expect(harness.mock.sent.length).toBeGreaterThan(before)); + const request = parseSentFrame(harness.mock.sent.at(-1)); + harness.mock.triggerMessage( + wrapSessionMessage({ + type: "plugin.catalog.get.response", + payload: { requestId: request.requestId, plugins: respond(request) }, + }), + ); + return { request, plugins: await response }; + } + + test("reuses retained client bundles the daemon leaves out", async () => { + const harness = await connectedClient(); + const first = await fetchCatalog(harness, () => [ + { id: "kept", clientBundle: "kept bundle", clientBundleHash: "hash-kept" }, + { id: "changed", clientBundle: "old bundle", clientBundleHash: "hash-old" }, + ]); + expect(first.request).not.toHaveProperty("cachedClientBundles"); + + const second = await fetchCatalog(harness, () => [ + { id: "kept", clientBundleHash: "hash-kept" }, + { id: "changed", clientBundle: "new bundle", clientBundleHash: "hash-new" }, + ]); + + expect(second.request.cachedClientBundles).toEqual([ + { id: "kept", clientBundleHash: "hash-kept" }, + { id: "changed", clientBundleHash: "hash-old" }, + ]); + expect(second.plugins).toEqual([ + { id: "kept", clientBundle: "kept bundle", clientBundleHash: "hash-kept" }, + { id: "changed", clientBundle: "new bundle", clientBundleHash: "hash-new" }, + ]); + const third = await fetchCatalog(harness, () => []); + expect(third.request.cachedClientBundles).toEqual([ + { id: "kept", clientBundleHash: "hash-kept" }, + { id: "changed", clientBundleHash: "hash-new" }, + ]); + const fourth = await fetchCatalog(harness, () => []); + expect(fourth.request).not.toHaveProperty("cachedClientBundles"); + }); + + test("keeps fetching every bundle from a daemon that sends no bundle hashes", async () => { + const harness = await connectedClient(); + const legacyCatalog = () => [{ id: "legacy", clientBundle: "legacy bundle" }]; + await fetchCatalog(harness, legacyCatalog); + + const second = await fetchCatalog(harness, legacyCatalog); + + expect(second.request).not.toHaveProperty("cachedClientBundles"); + expect(second.plugins).toEqual([{ id: "legacy", clientBundle: "legacy bundle" }]); + }); + + test("a fresh client advertises persisted bundles on its first fetch and expands omitted code", async () => { + let persisted: PluginClientBundle[] = []; + let loads = 0; + const store: PluginClientBundleStore = { + async load() { + loads++; + return persisted; + }, + async replace(bundles) { + persisted = bundles; + }, + }; + const createdFor: string[] = []; + const createStore = (serverId: string) => { + createdFor.push(serverId); + return store; + }; + const bundle = { id: "kept", clientBundleHash: "hash-kept", clientBundle: "stored code" }; + const warm = await connectedClient(createStore); + await fetchCatalog(warm, () => [bundle]); + const cold = await connectedClient(createStore); + const first = await fetchCatalog(cold, () => [{ id: "kept", clientBundleHash: "hash-kept" }]); + expect(first.request.cachedClientBundles).toEqual([ + { id: "kept", clientBundleHash: "hash-kept" }, + ]); + expect(first.plugins).toEqual([bundle]); + expect(loads).toBe(2); + await fetchCatalog(cold, () => []); + expect(loads).toBe(2); + expect(createdFor).toEqual([ + warm.client.getLastServerInfoMessage()!.serverId, + cold.client.getLastServerInfoMessage()!.serverId, + ]); + expect(persisted).toEqual([]); + }); + + test.each(["missing", "unreadable"])( + "a %s persistent cache requests full bundles", + async (state) => { + const store: PluginClientBundleStore = { + async load() { + if (state === "unreadable") throw new Error("Storage unavailable"); + return []; + }, + async replace() {}, + }; + const bundle = { id: "full", clientBundleHash: "hash-full", clientBundle: "full code" }; + const result = await fetchCatalog(await connectedClient(() => store), () => [bundle]); + expect(result.request).not.toHaveProperty("cachedClientBundles"); + expect(result.plugins).toEqual([bundle]); + }, + ); + + test("failed cache writes never fail plugin loading", async () => { + const store: PluginClientBundleStore = { + async load() { + return []; + }, + async replace() { + throw new Error("Disk full"); + }, + }; + const bundle = { id: "full", clientBundleHash: "hash-full", clientBundle: "full code" }; + expect( + (await fetchCatalog(await connectedClient(() => store), () => [bundle])).plugins, + ).toEqual([bundle]); + }); + + test.each([false, true])( + "usable catalogs do not wait for storage; latest write survives a failed first write: %s", + async (failFirstWrite) => { + const writes: PluginClientBundle[][] = []; + let finishWrite = () => {}; + const blockedWrite = new Promise((resolve) => { + finishWrite = resolve; + }); + const store: PluginClientBundleStore = { + async load() { + return []; + }, + async replace(bundles) { + writes.push(bundles); + if (writes.length === 1) { + await blockedWrite; + if (failFirstWrite) throw new Error("Disk full"); + } + }, + }; + const harness = await connectedClient(() => store); + const first = { id: "plugin", clientBundle: "old", clientBundleHash: "hash-old" }; + const changed = { id: "plugin", clientBundle: "new", clientBundleHash: "hash-new" }; + try { + expect((await fetchCatalog(harness, () => [first])).plugins).toEqual([first]); + await vi.waitFor(() => expect(writes).toEqual([[first]])); + expect((await fetchCatalog(harness, () => [changed])).plugins).toEqual([changed]); + expect((await fetchCatalog(harness, () => [])).plugins).toEqual([]); + expect(writes).toEqual([[first]]); + finishWrite(); + await vi.waitFor(() => expect(writes).toEqual([[first], []])); + } finally { + finishWrite(); + } + }, + ); +}); + test("keeps a plugin reload pending past the session RPC deadline", async () => { useHeartbeatClock(); try { diff --git a/packages/client/src/daemon-client.ts b/packages/client/src/daemon-client.ts index 524c3e264a..bca3d00301 100644 --- a/packages/client/src/daemon-client.ts +++ b/packages/client/src/daemon-client.ts @@ -396,6 +396,17 @@ export type DaemonEventHandler = (event: DaemonEvent) => void; export type BrowserAutomationExecuteRequestMessage = BrowserAutomationExecuteRequest; export type BrowserAutomationExecuteResponseMessage = BrowserAutomationExecuteResponse; +export interface PluginClientBundle { + id: string; + clientBundleHash: string; + clientBundle: string; +} + +/** Host-scoped storage. load must return only bundles whose contents have been verified. */ +export interface PluginClientBundleStore { + load(): Promise; + replace(bundles: PluginClientBundle[]): Promise; +} /** Endpoint security negotiated on a relay connection. See docs/relay-endpoint-security.md. */ export type RelaySecurityMode = "legacy" | "protected"; @@ -406,6 +417,8 @@ export interface DaemonClientConfig { sessionPermissions?: import("@getpaseo/protocol/messages").DaemonPermission[]; url: string; clientId: string; + /** Create one cache for the confirmed daemon identity, retained for this client lifetime. */ + createPluginClientBundleStore?: (serverId: string) => PluginClientBundleStore; clientType?: "mobile" | "browser" | "cli" | "mcp" | "hub"; appVersion?: string; runtimeGeneration?: number | null; @@ -1016,6 +1029,12 @@ type CorrelatedResponsePayloads = { type CorrelatedResponsePayload = CorrelatedResponsePayloads[TType]; +type PluginCatalogResponseEntry = + CorrelatedResponsePayload<"plugin.catalog.get.response">["plugins"][number]; +export type PluginCatalogEntry = Omit & { + clientBundle: string; +}; + export class DaemonConnectionError extends Error { readonly serverId?: string; readonly outcome?: "unknown"; @@ -1343,6 +1362,15 @@ export class DaemonClient { private pendingBinaryFileReads = new Map(); private activeBinaryFileTransfers = new Map(); private completedBinaryFileReads = new Map(); + /** The last catalog's client bundles by plugin id, so later catalogs can leave them out. */ + private pluginClientBundles = new Map< + string, + { clientBundleHash: string; clientBundle: string } + >(); + private pluginClientBundleStore: PluginClientBundleStore | undefined; + private pluginClientBundlesLoaded: Promise | undefined; + private pluginClientBundlesWrite: PluginClientBundle[] | undefined; + private pluginClientBundlesWriting = false; private logger: Logger; private pendingSendQueue: PendingSend[] = []; /** @@ -5793,14 +5821,82 @@ export class DaemonClient { }); } - async getPluginCatalog() { + async getPluginCatalog(): Promise { + if (this.config.createPluginClientBundleStore && !this.pluginClientBundleStore) { + const serverInfo = this.getLastServerInfoMessage(); + if (!serverInfo) throw new Error("Plugin catalog requires a connected host"); + this.pluginClientBundleStore = this.config.createPluginClientBundleStore(serverInfo.serverId); + } + const store = this.pluginClientBundleStore; + if (store) { + this.pluginClientBundlesLoaded ??= store.load().then( + (bundles) => { + for (const { id, ...bundle } of bundles) this.pluginClientBundles.set(id, bundle); + return undefined; + }, + (error) => this.logger.warn({ err: error }, "Failed to load plugin client bundles"), + ); + await this.pluginClientBundlesLoaded; + } const requestId = this.createRequestId(); + const cached = this.pluginClientBundles; + const cachedClientBundles = [...cached].map(([id, { clientBundleHash }]) => ({ + id, + clientBundleHash, + })); const payload = await this.sendCorrelatedSessionRequest({ requestId, - message: { type: "plugin.catalog.get.request", requestId }, + message: { + type: "plugin.catalog.get.request", + requestId, + ...(cachedClientBundles.length > 0 ? { cachedClientBundles } : {}), + }, responseType: "plugin.catalog.get.response", }); - return payload.plugins; + const retained = new Map(); + const plugins = payload.plugins.map(({ clientBundle: sent, ...plugin }): PluginCatalogEntry => { + const previous = cached.get(plugin.id); + const clientBundle = + sent ?? + (previous?.clientBundleHash === plugin.clientBundleHash + ? previous?.clientBundle + : undefined); + if (clientBundle === undefined) { + throw new Error(`Plugin catalog left out the client bundle for ${plugin.id}`); + } + // COMPAT(pluginCatalogBundleHash): introduced by PR #6732 (unreleased); remove after + // 2027-04-27 once the daemon floor includes bundle hashes. Older bundles are not retained. + if (plugin.clientBundleHash !== undefined) { + retained.set(plugin.id, { clientBundleHash: plugin.clientBundleHash, clientBundle }); + } + return Object.assign(plugin, { clientBundle }); + }); + this.pluginClientBundles = retained; + if (store) { + this.pluginClientBundlesWrite = [...retained].map(([id, bundle]) => + Object.assign({ id }, bundle), + ); + void this.persistPluginClientBundles(store); + } + return plugins; + } + + private async persistPluginClientBundles(store: PluginClientBundleStore): Promise { + if (this.pluginClientBundlesWriting) return; + this.pluginClientBundlesWriting = true; + try { + while (this.pluginClientBundlesWrite) { + const bundles = this.pluginClientBundlesWrite; + this.pluginClientBundlesWrite = undefined; + try { + await store.replace(bundles); + } catch (error) { + this.logger.warn({ err: error }, "Failed to persist plugin client bundles"); + } + } + } finally { + this.pluginClientBundlesWriting = false; + } } async listPlugins(): Promise { diff --git a/packages/protocol/src/messages.plugins.test.ts b/packages/protocol/src/messages.plugins.test.ts index f266b65988..47e91493fe 100644 --- a/packages/protocol/src/messages.plugins.test.ts +++ b/packages/protocol/src/messages.plugins.test.ts @@ -23,6 +23,29 @@ describe("plugin protocol compatibility", () => { expect(SessionOutboundMessageSchema.parse(message)).toEqual(message); }, ); + it.each([ + { type: "plugin.catalog.get.request", requestId: "catalog" }, + { + type: "plugin.catalog.get.request", + requestId: "catalog", + cachedClientBundles: [{ id: "example", clientBundleHash: "a".repeat(64) }], + }, + ])("parses plugin catalog requests %j", (request) => { + expect(SessionInboundMessageSchema.parse(request)).toEqual(request); + }); + it("parses plugin catalogs that leave out a client bundle the client already holds", () => { + const message = { + type: "plugin.catalog.get.response", + payload: { + requestId: "catalog", + plugins: [ + { id: "cached", clientBundleHash: "a".repeat(64) }, + { id: "changed", clientBundle: "bundle", clientBundleHash: "b".repeat(64) }, + ], + }, + }; + expect(SessionOutboundMessageSchema.parse(message)).toEqual(message); + }); it("preserves manifest metadata in plugin list responses", () => { const message = { type: "plugin.list.response", diff --git a/packages/protocol/src/messages.ts b/packages/protocol/src/messages.ts index 704c7ad277..92d6ec71e1 100644 --- a/packages/protocol/src/messages.ts +++ b/packages/protocol/src/messages.ts @@ -1470,6 +1470,10 @@ export const DiagnosticsRequestSchema = z.object({ export const PluginCatalogGetRequestSchema = z.object({ type: z.literal("plugin.catalog.get.request"), requestId: z.string(), + /** Client bundles the client already holds; the daemon leaves out each one whose hash matches. */ + cachedClientBundles: z + .array(z.object({ id: PluginIdSchema, clientBundleHash: z.string() })) + .optional(), }); export const PluginListRequestSchema = z.object({ @@ -6757,7 +6761,10 @@ export const PluginCatalogGetResponseSchema = z.object({ plugins: z.array( z.object({ id: PluginIdSchema, - clientBundle: z.string(), + /** Left out only when the request listed this plugin with the same `clientBundleHash`. */ + clientBundle: z.string().optional(), + /** SHA-256 of `clientBundle`, hex encoded. */ + clientBundleHash: z.string().optional(), requirements: PluginRequirementsSchema.optional(), }), ), diff --git a/packages/server/src/server/plugins/builtin/index.test.ts b/packages/server/src/server/plugins/builtin/index.test.ts index 6160c8f855..5f4728ecfa 100644 --- a/packages/server/src/server/plugins/builtin/index.test.ts +++ b/packages/server/src/server/plugins/builtin/index.test.ts @@ -78,6 +78,45 @@ test("listed client bundle is published while plugins are disabled; unlisted dir } }, 60_000); +test("a repeated catalog fetch leaves out the client bundle the client already holds", async () => { + const root = await mkdtemp(path.join(os.tmpdir(), "paseo-builtin-catalog-cache-")); + roots.push(root); + const directory = await fixture(root, "large", true); + // About the size of a real plugin client bundle. + await writeFile( + path.join(directory, "index.client.tsx"), + `const payload = ${JSON.stringify("x".repeat(1_000_000))}; +export default function contribute() { void payload; return () => {}; }`, + ); + const daemon = await createTestPaseoDaemon({ + daemonVersion: "0.9.2", + pluginsEnabled: false, + builtinPlugins: new BuiltinPluginLoader(root, ["large"]), + }); + const client = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws`, appVersion: "0.9.2" }); + const responseBytes: number[] = []; + const unsubscribe = client.subscribeRawMessages((message) => { + if (message.type === "plugin.catalog.get.response") { + responseBytes.push(JSON.stringify(message).length); + } + }); + try { + await client.connect(); + const first = await client.getPluginCatalog(); + const second = await client.getPluginCatalog(); + + expect(second).toEqual(first); + expect(second[0]?.clientBundle).toContain("x".repeat(1_000_000)); + expect(responseBytes).toHaveLength(2); + expect(responseBytes[0]).toBeGreaterThan(1_000_000); + expect(responseBytes[1]).toBeLessThan(500); + } finally { + unsubscribe(); + await client.close(); + await daemon.close(); + } +}, 60_000); + test("directory, Git, and npm installs reject a built-in ID", async () => { const root = await mkdtemp(path.join(os.tmpdir(), "paseo-builtin-install-")); roots.push(root); diff --git a/packages/server/src/server/plugins/requirements.posix.test.ts b/packages/server/src/server/plugins/requirements.posix.test.ts index 65462336a3..597d907e18 100644 --- a/packages/server/src/server/plugins/requirements.posix.test.ts +++ b/packages/server/src/server/plugins/requirements.posix.test.ts @@ -1,4 +1,5 @@ import { mkdtemp, readFile, readdir, rm, writeFile } from "node:fs/promises"; +import { createHash } from "node:crypto"; import { tmpdir } from "node:os"; import path from "node:path"; import { pathToFileURL } from "node:url"; @@ -68,9 +69,18 @@ it("rejects incompatible installs without persisting them, then accepts the corr await expect(service.installDirectory({ path: root })).resolves.toMatchObject({ status: "running", }); - expect(service.catalog()).toEqual([ - { id: "example", requirements: { paseo: "^0.8.0" }, clientBundle: expect.any(String) }, + const catalog = service.catalog(); + expect(catalog).toEqual([ + { + id: "example", + requirements: { paseo: "^0.8.0" }, + clientBundle: expect.any(String), + clientBundleHash: expect.any(String), + }, ]); + expect(catalog[0]!.clientBundleHash).toBe( + createHash("sha256").update(catalog[0]!.clientBundle).digest("hex"), + ); }); it("marks pre-0.8 plugins failed on startup and recovers after migration and reload", async () => { diff --git a/packages/server/src/server/plugins/runtime.posix.test.ts b/packages/server/src/server/plugins/runtime.posix.test.ts index a777ff188e..8004fd52bf 100644 --- a/packages/server/src/server/plugins/runtime.posix.test.ts +++ b/packages/server/src/server/plugins/runtime.posix.test.ts @@ -1,5 +1,6 @@ import { mkdir, mkdtemp, readdir, readFile, rm, writeFile } from "node:fs/promises"; import { fork } from "node:child_process"; +import { createHash } from "node:crypto"; import { tmpdir } from "node:os"; import { PassThrough } from "node:stream"; import path from "node:path"; @@ -106,6 +107,12 @@ function createReloadChild( }; } +function sha256(text: string | undefined): string { + return createHash("sha256") + .update(text ?? "") + .digest("hex"); +} + function createTestRuntime( dependencies: NonNullable[2]> = {}, logger = pino({ level: "silent" }), @@ -1536,6 +1543,7 @@ export default function contribute(server: { registerProvider(provider: Provider expect(spawnChild).not.toHaveBeenCalled(); expect(runtime.catalog()[0]?.clientBundle).toContain("addTheme"); + expect(runtime.catalog()[0]?.clientBundleHash).toBe(sha256(runtime.catalog()[0]?.clientBundle)); await runtime.stopAll(); }); @@ -1698,6 +1706,7 @@ export function inspectHost(_input: z.input) { const plugin = runtime.catalog()[0]; expect(plugin?.clientBundle).toContain("Client-only surface"); expect(plugin?.clientBundle).not.toContain("node:os"); + expect(plugin?.clientBundleHash).toBe(sha256(plugin?.clientBundle)); await expect(runtime.invoke("split-runtime", "inspect", {})).resolves.toMatchObject({ platform: expect.any(String), }); diff --git a/packages/server/src/server/plugins/runtime.ts b/packages/server/src/server/plugins/runtime.ts index 182575bf80..5a15020cef 100644 --- a/packages/server/src/server/plugins/runtime.ts +++ b/packages/server/src/server/plugins/runtime.ts @@ -15,7 +15,7 @@ import { validateBeforeRequest, validateBeforeResult } from "./lifecycle/index.j import { fork } from "node:child_process"; import path from "node:path"; import { fileURLToPath } from "node:url"; -import { randomUUID } from "node:crypto"; +import { createHash, randomUUID } from "node:crypto"; import type pino from "pino"; import { PROVIDER_CAPABILITIES, @@ -74,6 +74,8 @@ interface LoadedPlugin { instanceId: string; requirements: PluginRequirements | undefined; clientBundle: string; + /** Lets a client that already holds this bundle skip downloading it again. */ + clientBundleHash: string; methods: ReadonlySet; hooks: { events: string[]; before: string[] }; providers: readonly PluginProviderMetadata[]; @@ -246,6 +248,10 @@ function terminatePluginChild(child: PluginChild): void { if (!child.killed) child.kill(); } +function hashClientBundle(clientBundle: string): string { + return createHash("sha256").update(clientBundle).digest("hex"); +} + function describeError(error: unknown): string { return error instanceof Error ? error.message : String(error); } @@ -369,9 +375,19 @@ export class PluginRuntime { return true; } - catalog(): Array<{ id: string; clientBundle: string; requirements?: PluginRequirements }> { + catalog(): Array<{ + id: string; + clientBundle: string; + clientBundleHash: string; + requirements?: PluginRequirements; + }> { return [...this.plugins.values()] - .map(({ id, clientBundle, requirements }) => ({ id, clientBundle, requirements })) + .map(({ id, clientBundle, clientBundleHash, requirements }) => ({ + id, + clientBundle, + clientBundleHash, + requirements, + })) .sort((left, right) => left.id.localeCompare(right.id)); } @@ -739,10 +755,12 @@ export class PluginRuntime { if (!canPublish()) throw new Error(`Plugin start cancelled: ${pluginId}`); const serverBundle = bundles.serverBundle; if (!serverBundle) { + const clientBundle = bundles.clientBundle ?? ""; return { id: pluginId, instanceId: randomUUID(), - clientBundle: bundles.clientBundle ?? "", + clientBundle, + clientBundleHash: hashClientBundle(clientBundle), requirements: manifest.requirements, methods: new Set(), hooks: { events: [], before: [] }, @@ -872,6 +890,7 @@ export class PluginRuntime { id: pluginId, instanceId: randomUUID(), clientBundle, + clientBundleHash: hashClientBundle(clientBundle), requirements, methods: new Set(ready.methods), hooks: ready.hooks ?? { events: [], before: [] }, diff --git a/packages/server/src/server/session.test.ts b/packages/server/src/server/session.test.ts index 7321312826..3cdef95a83 100644 --- a/packages/server/src/server/session.test.ts +++ b/packages/server/src/server/session.test.ts @@ -588,6 +588,78 @@ test("routes plugin requests and releases its owned catalog subscription on clea expect(releasePluginSubscription).toHaveBeenCalledOnce(); }); +test("leaves out the client bundles a catalog request already holds", async () => { + const messages: SessionOutboundMessage[] = []; + const pluginRuntime: NonNullable = { + before: async (_name, request) => request, + emit: () => {}, + listPlugins: async () => [], + getLogs: () => [], + installDirectory: async () => { + throw new Error("Unexpected install"); + }, + inspectDirectory: async () => ({ id: "unused" }), + reloadPlugin: async () => { + throw new Error("Unexpected reload"); + }, + enablePlugin: async () => { + throw new Error("Unexpected enable"); + }, + disablePlugin: async () => { + throw new Error("Unexpected disable"); + }, + removePlugin: async () => undefined, + subscribe: () => () => {}, + catalog: () => [ + { id: "cached", clientBundle: "cached bundle", clientBundleHash: "hash-cached" }, + { id: "changed", clientBundle: "new bundle", clientBundleHash: "hash-new" }, + { id: "added", clientBundle: "added bundle", clientBundleHash: "hash-added" }, + ], + invokePluginRpc: async () => null, + listUsageReports: async () => [], + listLegacyUsage: async () => ({ fetchedAt: "2026-01-01T00:00:00.000Z", providers: [] }), + }; + const session = createSessionForTest({ messages, pluginRuntime }); + + await session.handleMessage({ + type: "plugin.catalog.get.request", + requestId: "reconnect", + cachedClientBundles: [ + { id: "cached", clientBundleHash: "hash-cached" }, + { id: "changed", clientBundleHash: "hash-old" }, + { id: "removed", clientBundleHash: "hash-removed" }, + ], + }); + // Clients from before cachedClientBundles always receive every bundle. + await session.handleMessage({ type: "plugin.catalog.get.request", requestId: "legacy" }); + + expect(messages).toEqual([ + { + type: "plugin.catalog.get.response", + payload: { + requestId: "reconnect", + plugins: [ + { id: "cached", clientBundleHash: "hash-cached" }, + { id: "changed", clientBundle: "new bundle", clientBundleHash: "hash-new" }, + { id: "added", clientBundle: "added bundle", clientBundleHash: "hash-added" }, + ], + }, + }, + { + type: "plugin.catalog.get.response", + payload: { + requestId: "legacy", + plugins: [ + { id: "cached", clientBundle: "cached bundle", clientBundleHash: "hash-cached" }, + { id: "changed", clientBundle: "new bundle", clientBundleHash: "hash-new" }, + { id: "added", clientBundle: "added bundle", clientBundleHash: "hash-added" }, + ], + }, + }, + ]); + await session.cleanup(); +}); + describe("workspace label subscriptions", () => { type LabelSubscription = Awaited>; type LabelChange = Parameters[0]["onChange"]>[0]; diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 3e78982793..b52c9dbfe4 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -518,7 +518,7 @@ export interface SessionOptions { removePlugin(pluginId: string): Promise; subscribe(listener: (pluginId: string) => void): () => void; subscribeSettings?(listener: (pluginId: string, settingsId: string) => void): () => void; - catalog(): Array<{ id: string; clientBundle: string }>; + catalog(): Array<{ id: string; clientBundle: string; clientBundleHash?: string }>; invokePluginRpc(pluginId: string, method: string, input: unknown): Promise; listUsageReports(options?: { forceRefresh?: boolean; @@ -2584,12 +2584,17 @@ export class Session { }); } if (msg.type === "plugin.catalog.get.request") { + const cached = new Map( + msg.cachedClientBundles?.map((entry) => [entry.id, entry.clientBundleHash]), + ); + const plugins = (this.pluginRuntime?.catalog() ?? []).map(({ clientBundle, ...plugin }) => + plugin.clientBundleHash !== undefined && cached.get(plugin.id) === plugin.clientBundleHash + ? plugin + : Object.assign(plugin, { clientBundle }), + ); this.emit({ type: "plugin.catalog.get.response", - payload: { - requestId: msg.requestId, - plugins: this.pluginRuntime?.catalog() ?? [], - }, + payload: { requestId: msg.requestId, plugins }, }); return undefined; }