mirror of
https://github.com/getpaseo/paseo.git
synced 2026-11-04 15:34:39 +00:00
Tahir Khan
b807c261ca
Plugin screens can report unavailable while a successful catalog request is still pending. Track the host catalog state so these screens show progress until their contributions register. Keep missing surfaces, failed requests, and offline hosts distinct from loading. Retain existing installed surfaces during catalog refreshes. Cover pending-to-ready rendering with a real browser regression and catalog lifecycle tests. Related to getpaseo/paseo#7088
505 lines
17 KiB
TypeScript
505 lines
17 KiB
TypeScript
import appPackage from "../../package.json";
|
|
import { afterEach, expect, it } from "vitest";
|
|
import { QueryObserver } from "@tanstack/react-query";
|
|
import { DaemonClient } from "@getpaseo/client/internal/daemon-client";
|
|
import type { SessionOutboundMessage } from "@getpaseo/protocol/messages";
|
|
import { settingsRpc } from "@getpaseo/plugin";
|
|
import { syncPluginCatalog } from "./catalog-sync-model";
|
|
import { createPluginHosts } from "./hosts";
|
|
import { PluginRegistry } from "./registry";
|
|
import { pluginSettingsKey } from "./settings/query";
|
|
import type { InstalledPlugin } from "./types";
|
|
|
|
const audio = { play: async () => 0 };
|
|
const rpc = settingsRpc("preferences");
|
|
type Catalog = Awaited<ReturnType<DaemonClient["getPluginCatalog"]>>;
|
|
type Observation = ReturnType<DaemonClient["observeEvents"]>;
|
|
type Observer = Parameters<Observation["subscribe"]>[0];
|
|
|
|
class SettingsStore {
|
|
private revision = 1;
|
|
private values = { label: "original" };
|
|
reads = 0;
|
|
|
|
read() {
|
|
this.reads++;
|
|
return { status: "ready", revision: String(this.revision), values: this.values } as const;
|
|
}
|
|
|
|
write(input: unknown) {
|
|
const write = rpc.write.input.parse(input);
|
|
if (write.revision !== String(this.revision)) {
|
|
return { status: "conflict", error: "Settings changed" } as const;
|
|
}
|
|
const values = write.values;
|
|
if (typeof values !== "object" || values === null || !("label" in values)) {
|
|
throw new Error("Expected settings values");
|
|
}
|
|
if (typeof values.label !== "string") throw new Error("Expected label");
|
|
this.values = { label: values.label };
|
|
this.revision++;
|
|
return { status: "saved", revision: String(this.revision), values: this.values } as const;
|
|
}
|
|
}
|
|
|
|
/** In-memory daemon port; it delivers owned snapshots and stops updates on release. */
|
|
class CatalogClient extends DaemonClient {
|
|
catalog = catalog("one");
|
|
fetches = 0;
|
|
releases = 0;
|
|
events: Parameters<DaemonClient["observeEvents"]>[0] = [];
|
|
private readonly observers = new Set<Observer>();
|
|
|
|
constructor(readonly settings = new SettingsStore()) {
|
|
super({ url: "ws://unused.test", clientId: "catalog-sync-test" });
|
|
}
|
|
|
|
override async getPluginCatalog(): Promise<Catalog> {
|
|
this.fetches++;
|
|
return this.catalog;
|
|
}
|
|
|
|
override observeEvents(events: Parameters<DaemonClient["observeEvents"]>[0]): Observation {
|
|
this.events = events;
|
|
const snapshot = { requestId: "events", subscriptionId: "catalog" };
|
|
const subscribers = new Set<Observer>();
|
|
return {
|
|
subscriptionId: snapshot.subscriptionId,
|
|
ready: Promise.resolve(snapshot),
|
|
subscribe: (observer) => {
|
|
subscribers.add(observer);
|
|
this.observers.add(observer);
|
|
observer.snapshot(snapshot);
|
|
return () => {
|
|
subscribers.delete(observer);
|
|
this.observers.delete(observer);
|
|
};
|
|
},
|
|
release: async () => {
|
|
this.releases++;
|
|
for (const observer of subscribers) this.observers.delete(observer);
|
|
subscribers.clear();
|
|
},
|
|
};
|
|
}
|
|
|
|
override async invokePluginRpc(_pluginId: string, method: string, input: unknown) {
|
|
if (method === rpc.read.name) return this.settings.read();
|
|
if (method === rpc.write.name) return this.settings.write(input);
|
|
throw new Error(`Unexpected RPC: ${method}`);
|
|
}
|
|
|
|
update(message: SessionOutboundMessage) {
|
|
for (const observer of this.observers) observer.update(message);
|
|
}
|
|
|
|
snapshot() {
|
|
for (const observer of this.observers) {
|
|
observer.snapshot({ requestId: "reconnect", subscriptionId: "catalog" });
|
|
}
|
|
}
|
|
}
|
|
|
|
function catalog(marker: string): Catalog {
|
|
return [
|
|
{
|
|
id: "reports",
|
|
requirements: { paseo: `>=${appPackage.version}` },
|
|
clientBundle: `(function() { return { default: function(plugin) {
|
|
plugin.addTimelineTransformer({
|
|
id: "report", query: { itemType: "tool_call" },
|
|
transform() { return ${JSON.stringify(marker)} ? { items: [] } : undefined; }
|
|
});
|
|
return function() { plugin.openSettings("cleanup"); };
|
|
} }; })`,
|
|
},
|
|
];
|
|
}
|
|
|
|
const cleanups: Array<() => void> = [];
|
|
afterEach(() => {
|
|
for (const cleanup of cleanups.splice(0).toReversed()) cleanup();
|
|
});
|
|
|
|
function harness() {
|
|
let evaluations = 0;
|
|
let disposals = 0;
|
|
const registry = new PluginRegistry({
|
|
localeSource: { getLocale: () => "en", subscribe: () => () => {} },
|
|
version: appPackage.version,
|
|
createRuntime(installation) {
|
|
evaluations++;
|
|
return {
|
|
hosts: createPluginHosts(
|
|
{
|
|
getHosts: () => [],
|
|
getSnapshot: () => null,
|
|
subscribeAll: () => () => {},
|
|
subscribeHostList: () => () => {},
|
|
},
|
|
installation.lifetime.signal,
|
|
),
|
|
paseo: installation.paseo,
|
|
i18n: installation.i18n,
|
|
rpc: async () => {
|
|
throw new Error("Unexpected runtime RPC");
|
|
},
|
|
openSettings: () => {
|
|
disposals++;
|
|
},
|
|
openScreen() {},
|
|
openSurface() {},
|
|
async playAudio() {},
|
|
openPanel() {},
|
|
addComposerPill: () => ({ update() {}, remove() {} }),
|
|
addHeaderButton: () => ({ update() {}, remove() {} }),
|
|
subscribe: () => {
|
|
throw new Error("Subscriptions are not used here");
|
|
},
|
|
};
|
|
},
|
|
});
|
|
const client = new CatalogClient();
|
|
cleanups.push(() => {
|
|
registry.removeHost("host");
|
|
registry.removeHost("other");
|
|
});
|
|
function sync(options: Partial<Parameters<typeof syncPluginCatalog>[0]> = {}) {
|
|
const stop = syncPluginCatalog({
|
|
serverId: "host",
|
|
client,
|
|
audio,
|
|
connected: true,
|
|
supported: true,
|
|
registry,
|
|
...options,
|
|
});
|
|
cleanups.push(stop);
|
|
return stop;
|
|
}
|
|
function installed(serverId = "host") {
|
|
const plugin = registry.getSnapshot().find((entry) => entry.serverId === serverId);
|
|
if (!plugin) throw new Error(`Missing installation: ${serverId}`);
|
|
return plugin;
|
|
}
|
|
return {
|
|
registry,
|
|
client,
|
|
sync,
|
|
installed,
|
|
evaluations: () => evaluations,
|
|
disposals: () => disposals,
|
|
};
|
|
}
|
|
|
|
async function connected() {
|
|
const h = harness();
|
|
const stop = h.sync();
|
|
await expect.poll(() => h.registry.getSnapshot().length).toBe(1);
|
|
return { ...h, stop };
|
|
}
|
|
|
|
it("keeps a restored plugin screen loading until its catalog arrives", async () => {
|
|
const h = harness();
|
|
expect(h.registry.getCatalogStatus("host")).toBe("loading");
|
|
const pending = Promise.withResolvers<Catalog>();
|
|
class PendingClient extends CatalogClient {
|
|
override getPluginCatalog() {
|
|
this.fetches++;
|
|
return pending.promise;
|
|
}
|
|
}
|
|
const client = new PendingClient();
|
|
h.sync({ client });
|
|
|
|
expect(h.registry.getCatalogStatus("host")).toBe("loading");
|
|
expect(h.registry.getSnapshot()).toEqual([]);
|
|
pending.resolve(client.catalog);
|
|
|
|
await expect.poll(() => h.registry.getCatalogStatus("host")).toBe("ready");
|
|
expect(h.installed().id).toBe("reports");
|
|
expect(client.fetches).toBe(1);
|
|
});
|
|
|
|
it("settles an empty catalog without leaving missing plugin screens loading", async () => {
|
|
const h = harness();
|
|
h.client.catalog = [];
|
|
const snapshot = h.registry.getSnapshot();
|
|
const statuses: string[] = [];
|
|
cleanups.push(
|
|
h.registry.subscribe(() => {
|
|
statuses.push(h.registry.getCatalogStatus("host"));
|
|
}),
|
|
);
|
|
h.sync();
|
|
|
|
await expect.poll(() => h.registry.getCatalogStatus("host")).toBe("ready");
|
|
expect(h.registry.getSnapshot()).toBe(snapshot);
|
|
expect(statuses).toEqual(["loading", "ready"]);
|
|
});
|
|
|
|
it("settles a failed catalog request instead of loading indefinitely", async () => {
|
|
const h = harness();
|
|
class FailingClient extends CatalogClient {
|
|
override async getPluginCatalog(): Promise<Catalog> {
|
|
throw new Error("Catalog request failed");
|
|
}
|
|
}
|
|
h.sync({ client: new FailingClient() });
|
|
|
|
await expect.poll(() => h.registry.getCatalogStatus("host")).toBe("error");
|
|
expect(h.registry.getSnapshot()).toEqual([]);
|
|
});
|
|
|
|
it("ignores an old failed request after a new connection loads its catalog", async () => {
|
|
const h = harness();
|
|
const pending = Promise.withResolvers<Catalog>();
|
|
const started = Promise.withResolvers<void>();
|
|
class PendingClient extends CatalogClient {
|
|
override getPluginCatalog() {
|
|
started.resolve();
|
|
return pending.promise;
|
|
}
|
|
}
|
|
const stop = h.sync({ client: new PendingClient() });
|
|
await started.promise;
|
|
stop();
|
|
h.sync();
|
|
await expect.poll(() => h.registry.getCatalogStatus("host")).toBe("ready");
|
|
const plugin = h.installed();
|
|
|
|
pending.reject(new Error("Old connection failed"));
|
|
await pending.promise.catch(() => undefined);
|
|
await new Promise<void>((resolve) => setTimeout(resolve, 0));
|
|
|
|
expect(h.registry.getCatalogStatus("host")).toBe("ready");
|
|
expect(h.installed()).toBe(plugin);
|
|
});
|
|
|
|
function settingsQuery(plugin: InstalledPlugin) {
|
|
return {
|
|
queryKey: pluginSettingsKey("preferences"),
|
|
queryFn: async () => rpc.read.output.parse(await plugin.invoke(rpc.read.name, {})),
|
|
// The replica policy used by useSettings deliberately disables automatic refetches.
|
|
gcTime: Infinity,
|
|
staleTime: Infinity,
|
|
refetchOnMount: false,
|
|
refetchOnReconnect: false,
|
|
refetchOnWindowFocus: false,
|
|
retry: false,
|
|
} as const;
|
|
}
|
|
|
|
it("retains installations and timeline transforms while disconnected", async () => {
|
|
const h = await connected();
|
|
const plugin = h.installed();
|
|
const snapshot = h.registry.getSnapshot();
|
|
const transformers = plugin.timelineTransformers;
|
|
h.stop();
|
|
h.sync({ connected: false });
|
|
h.client.update({
|
|
type: "status",
|
|
payload: { status: "plugin_catalog_changed", pluginId: "reports" },
|
|
});
|
|
expect(h.registry.getSnapshot()).toBe(snapshot);
|
|
expect(h.installed().timelineTransformers).toBe(transformers);
|
|
expect(plugin.lifetime.signal.aborted).toBe(false);
|
|
expect(h.disposals()).toBe(0);
|
|
expect(h.client.fetches).toBe(1);
|
|
expect(h.client.events).toEqual([
|
|
"status.plugin_catalog_changed",
|
|
"status.plugin_settings_changed",
|
|
]);
|
|
});
|
|
|
|
it.each(["mounted", "unmounted"])(
|
|
"refreshes %s settings and revisions after another client saves offline",
|
|
async (surface) => {
|
|
const h = await connected();
|
|
const plugin = h.installed();
|
|
const snapshot = h.registry.getSnapshot();
|
|
const options = settingsQuery(plugin);
|
|
const original = await plugin.queryClient.fetchQuery(options);
|
|
const observer = new QueryObserver(plugin.queryClient, options);
|
|
const unsubscribe = observer.subscribe(() => {});
|
|
cleanups.push(unsubscribe);
|
|
if (surface === "unmounted") unsubscribe();
|
|
plugin.queryClient.setQueryData(["attachment-search"], "retained");
|
|
const otherClient = new CatalogClient();
|
|
h.registry.installCatalog("other", otherClient.catalog, { client: otherClient, audio });
|
|
const otherPlugin = h.installed("other");
|
|
await otherPlugin.queryClient.fetchQuery(settingsQuery(otherPlugin));
|
|
// Another host publishes new presentation snapshots without replacing the owning lifetime.
|
|
const beforeReconnect = h.registry.getSnapshot();
|
|
const retainedInstallation = h.installed();
|
|
|
|
h.stop();
|
|
h.sync({ connected: false });
|
|
const clientB = new CatalogClient(h.client.settings);
|
|
expect(
|
|
await clientB.invokePluginRpc("reports", rpc.write.name, {
|
|
revision: original.revision,
|
|
values: { label: "client B" },
|
|
}),
|
|
).toEqual({ status: "saved", revision: "2", values: { label: "client B" } });
|
|
expect(plugin.queryClient.getQueryData(options.queryKey)).toEqual(original);
|
|
h.sync();
|
|
|
|
await expect
|
|
.poll(() => plugin.queryClient.getQueryData(options.queryKey))
|
|
.toEqual({
|
|
status: "ready",
|
|
revision: "2",
|
|
values: { label: "client B" },
|
|
});
|
|
const refreshed = await plugin.queryClient.fetchQuery(options);
|
|
expect(
|
|
await plugin.invoke(rpc.write.name, {
|
|
revision: refreshed.revision,
|
|
values: { label: "client A" },
|
|
}),
|
|
).toEqual({ status: "saved", revision: "3", values: { label: "client A" } });
|
|
expect(h.registry.getSnapshot()).toBe(beforeReconnect);
|
|
expect(h.installed()).toBe(retainedInstallation);
|
|
expect(h.installed().lifetime).toBe(plugin.lifetime);
|
|
expect(h.installed().queryClient).toBe(plugin.queryClient);
|
|
expect(h.installed().timelineTransformers).toBe(snapshot[0].timelineTransformers);
|
|
expect(plugin.queryClient.getQueryData(["attachment-search"])).toBe("retained");
|
|
expect(otherClient.settings.reads).toBe(1);
|
|
expect(otherPlugin.queryClient.getQueryState(options.queryKey)?.isInvalidated).toBe(false);
|
|
expect(otherPlugin.queryClient.getQueryData(options.queryKey)).toEqual(original);
|
|
expect(h.evaluations()).toBe(2);
|
|
expect(h.disposals()).toBe(0);
|
|
},
|
|
);
|
|
|
|
it("replaces changed bundles on reconnect and disposes the old installation", async () => {
|
|
const h = await connected();
|
|
const plugin = h.installed();
|
|
plugin.queryClient.setQueryData(["attachment-search"], "old");
|
|
h.stop();
|
|
h.client.catalog = catalog("two");
|
|
h.sync();
|
|
await expect.poll(() => h.installed() === plugin).toBe(false);
|
|
expect(plugin.lifetime.signal.aborted).toBe(true);
|
|
expect(plugin.queryClient.getQueryCache().getAll()).toEqual([]);
|
|
expect(h.installed().timelineTransformers.map((entry) => entry.id)).toEqual(["report"]);
|
|
expect(h.evaluations()).toBe(2);
|
|
expect(h.disposals()).toBe(1);
|
|
});
|
|
|
|
it("replaces an explicitly reloaded bundle even when its source is unchanged", async () => {
|
|
const h = await connected();
|
|
const plugin = h.installed();
|
|
h.client.update({
|
|
type: "status",
|
|
payload: { status: "plugin_catalog_changed", pluginId: "reports" },
|
|
});
|
|
await expect.poll(() => h.installed() === plugin).toBe(false);
|
|
expect(h.disposals()).toBe(1);
|
|
expect(h.evaluations()).toBe(2);
|
|
});
|
|
|
|
it.each(["unsupported", "no audio"])("disposes plugins when %s", async (reason) => {
|
|
const h = await connected();
|
|
const plugin = h.installed();
|
|
h.stop();
|
|
h.sync({ supported: reason !== "unsupported", audio: reason === "no audio" ? null : audio });
|
|
expect(h.registry.getSnapshot()).toEqual([]);
|
|
expect(plugin.lifetime.signal.aborted).toBe(true);
|
|
expect(h.disposals()).toBe(1);
|
|
});
|
|
|
|
it("refreshes settings on a replayed subscription snapshot without reinstalling", async () => {
|
|
const h = await connected();
|
|
const plugin = h.installed();
|
|
const options = settingsQuery(plugin);
|
|
await plugin.queryClient.fetchQuery(options);
|
|
h.client.settings.write({ revision: "1", values: { label: "replayed" } });
|
|
h.client.snapshot();
|
|
await expect
|
|
.poll(() => plugin.queryClient.getQueryData(options.queryKey))
|
|
.toEqual({
|
|
status: "ready",
|
|
revision: "2",
|
|
values: { label: "replayed" },
|
|
});
|
|
expect(h.installed()).toBe(plugin);
|
|
expect(h.evaluations()).toBe(1);
|
|
});
|
|
|
|
it("invalidates only the named settings on its host for a live notification", async () => {
|
|
const h = await connected();
|
|
const plugin = h.installed();
|
|
const options = settingsQuery(plugin);
|
|
await plugin.queryClient.fetchQuery(options);
|
|
const observer = new QueryObserver(plugin.queryClient, options);
|
|
cleanups.push(observer.subscribe(() => {}));
|
|
plugin.queryClient.setQueryData(pluginSettingsKey("other-settings"), "unchanged");
|
|
const other = new CatalogClient();
|
|
h.registry.installCatalog("other", other.catalog, { client: other, audio });
|
|
await h.installed("other").queryClient.fetchQuery(settingsQuery(h.installed("other")));
|
|
|
|
const beforeNotification = h.registry.getSnapshot();
|
|
const retainedInstallation = h.installed();
|
|
h.client.settings.write({ revision: "1", values: { label: "live" } });
|
|
h.client.update({
|
|
type: "status",
|
|
payload: {
|
|
status: "plugin_settings_changed",
|
|
pluginId: "reports",
|
|
settingsId: "preferences",
|
|
},
|
|
});
|
|
|
|
await expect
|
|
.poll(() => plugin.queryClient.getQueryData(options.queryKey))
|
|
.toEqual({
|
|
status: "ready",
|
|
revision: "2",
|
|
values: { label: "live" },
|
|
});
|
|
expect(plugin.queryClient.getQueryState(pluginSettingsKey("other-settings"))?.isInvalidated).toBe(
|
|
false,
|
|
);
|
|
expect(other.settings.reads).toBe(1);
|
|
expect(h.registry.getSnapshot()).toBe(beforeNotification);
|
|
expect(h.installed()).toBe(retainedInstallation);
|
|
expect(h.installed().lifetime).toBe(plugin.lifetime);
|
|
expect(h.installed().queryClient).toBe(plugin.queryClient);
|
|
expect(h.evaluations()).toBe(2);
|
|
expect(h.client.fetches).toBe(1);
|
|
});
|
|
|
|
it("ignores an in-flight catalog snapshot after disconnect", async () => {
|
|
const h = await connected();
|
|
const plugin = h.installed();
|
|
await plugin.queryClient.fetchQuery(settingsQuery(plugin));
|
|
const snapshot = h.registry.getSnapshot();
|
|
h.stop();
|
|
let resolveCatalog: (value: Catalog) => void = () => {};
|
|
const pending = new Promise<Catalog>((resolve) => {
|
|
resolveCatalog = resolve;
|
|
});
|
|
let fetching: () => void = () => {};
|
|
const started = new Promise<void>((resolve) => {
|
|
fetching = resolve;
|
|
});
|
|
class PendingClient extends CatalogClient {
|
|
override getPluginCatalog() {
|
|
fetching();
|
|
return pending;
|
|
}
|
|
}
|
|
const stop = h.sync({ client: new PendingClient(h.client.settings) });
|
|
await started;
|
|
stop();
|
|
resolveCatalog(catalog("late"));
|
|
await pending;
|
|
// Drain the catalog continuation through a subsequently queued task.
|
|
await new Promise<void>((resolve) => setTimeout(resolve, 0));
|
|
expect(h.registry.getSnapshot()).toBe(snapshot);
|
|
expect(h.client.settings.reads).toBe(1);
|
|
expect(h.evaluations()).toBe(1);
|
|
});
|