fix(plugins): expose server settings access (#4674)

* fix(plugins): expose server settings access

* fix(plugins): isolate settings subscriber state

* test(plugins): read persisted settings before clients connect

* test(app): reconnect seed client after old daemon restart

* test(server): drain file creation events before deletion assertions

* test(creation): expose errors when completion fails

---------

Co-authored-by: Mohamed Boudra <boudra.moha@gmail.com>
This commit is contained in:
Matt Cowger
2026-09-14 23:03:20 +02:00
committed by GitHub
co-authored by Mohamed Boudra
parent 13979e9c38
commit 0c5f472da1
12 changed files with 325 additions and 47 deletions
+4
View File
@@ -477,6 +477,10 @@ Its writer lives with the plugin subprocess, while its directory lives outside m
so updates and reloads retain values. Settings-change notifications must not enter the catalog
reload path: that path disposes the plugin and would destroy open drafts after every save.
`server.registerSettings(definition)` returns a server-side handle. Use `read()` for the current
`ready` or `invalid` state and `subscribe()` for successful saves, resets, and migrations. The
subscription cleanup belongs in the plugin's contribution cleanup when it outlives the entry.
## Contribute a theme
`addTheme` takes a small light or dark palette and a display name. Paseo expands it through the
@@ -113,6 +113,7 @@ test("does not repeat an assistant block when the current app paginates a publis
page,
"compatibility-after-reconnect: emit 1 coalesced agent stream updates",
);
// The browser and the seed client reconnect independently after the restart.
await agent.client.connect();
await agent.client.waitForFinish(agent.agentId, 20_000);
await page.screenshot({ path: test.info().outputPath("old-daemon-live-reconnect.png") });
+18 -1
View File
@@ -9,10 +9,27 @@ export interface PluginHandlerContext {
paseo: PaseoApi;
}
export type PluginSettingsState<Schema extends ZodType> =
| {
status: "ready";
revision: string;
values: ZodOutput<Schema>;
}
| {
status: "invalid";
revision: string;
error: string;
};
export interface PluginSettings<Schema extends ZodType> {
read(): Promise<PluginSettingsState<Schema>>;
subscribe(listener: (state: PluginSettingsState<Schema>) => void | Promise<void>): PluginCleanup;
}
export interface PluginServerContext extends PluginLifecycleRegistration {
registerSettings<Schema extends ZodType>(
definition: import("../settings.js").SettingsDefinition<Schema>,
): void;
): PluginSettings<Schema>;
handle<InputSchema extends ZodType, OutputSchema extends ZodType>(
contract: PluginRpcContract<InputSchema, OutputSchema>,
handler: (
+2
View File
@@ -2,6 +2,8 @@ export type {
PluginHandlerContext,
PluginServerContext,
PluginServerContribution,
PluginSettings,
PluginSettingsState,
} from "./contracts.js";
export type {
PluginHookContext,
@@ -113,7 +113,7 @@ test("workspace readiness is observable before provider startup and duplicate re
expect(f.updates[0]).toMatchObject({ workspaceId: workspace.id, agentId: agent.id });
const duplicate = f.service.create(f.input);
f.provider.resolve();
expect(await first).toMatchObject({ phase: "completed", workspace, agent });
expect(await first).toMatchObject({ phase: "completed", error: null, workspace, agent });
expect(await duplicate).toEqual(await first);
expect(f.calls).toEqual(["workspace", "agent", "prompt"]);
expect(await new CreationService(f.directory, silentLogger).create(f.input)).toEqual(await first);
@@ -213,7 +213,8 @@ test("observes a thousand concurrent writes and remains healthy after delete and
const directories = Array.from({ length: 20 }, (_, index) => join(root, `dir-${index}`));
await Promise.all(directories.map((directory) => mkdir(directory)));
const observed = new Map<string, Set<FileChange["type"]>>();
const subscription = await subscribeToFileChanges(root, (error, events) => {
const observer = createObserver();
const subscription = await observer.subscribe(root, (error, events) => {
expect(error).toBeNull();
for (const event of events) {
const types = observed.get(event.path) ?? new Set();
@@ -240,6 +241,16 @@ test("observes a thousand concurrent writes and remains healthy after delete and
.poll(() => paths.filter((path) => !observed.has(path)), { timeout: 60_000 })
.toEqual([]);
// Native notifications arrive before reconciliation finishes. Drain that work
// so pending creates cannot coalesce with this test's separate deletion phase.
await expect
.poll(() => observer.getDiagnostics(), { timeout: 10_000 })
.toMatchObject({
pendingEventCount: 0,
pendingReconciliationWorkCount: 0,
reconciliationInFlightCount: 0,
});
const removedPaths = paths.slice(0, 100);
await Promise.all(removedPaths.map((path) => rm(path)));
await expect
@@ -10,6 +10,7 @@ import * as pluginProviderRuntime from "@getpaseo/plugin/server/provider";
import * as pluginAcpRuntime from "@getpaseo/plugin/server/acp";
import type { SettingsDefinition, PluginRpcContract } from "@getpaseo/plugin";
import type { PluginHandlerContext } from "@getpaseo/plugin/server";
import type { ZodType } from "zod";
import {
ProviderEventSchema,
type ProviderConnection,
@@ -23,7 +24,7 @@ import { createPluginClientId } from "./plugin-session-identity.js";
import { PluginSettingsStore } from "./settings/index.js";
let settingsStore: PluginSettingsStore | null = null;
function registerSettings(definition: SettingsDefinition) {
function registerSettings<Schema extends ZodType>(definition: SettingsDefinition<Schema>) {
if (!settingsStore) throw new Error("Plugin settings storage is unavailable");
const handlers = settingsStore.register(definition);
register(handlers.read.contract, handlers.read.handle);
@@ -33,6 +34,7 @@ function registerSettings(definition: SettingsDefinition) {
register(handlers.reset.contract, (input) =>
handlers.reset.handle(handlers.reset.contract.input.parse(input)),
);
return handlers.settings;
}
type RpcHandler = (input: unknown, context: PluginHandlerContext) => unknown | Promise<unknown>;
@@ -1,19 +1,102 @@
import { resolveDaemonVersion } from "../daemon-version.js";
import { mkdtemp, rm, writeFile } from "node:fs/promises";
import { mkdir, mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
import { tmpdir } from "node:os";
import path from "node:path";
import { expect, test } from "vitest";
import { expect, onTestFinished, test } from "vitest";
import { z } from "zod";
import { settingsRpc } from "@getpaseo/plugin";
import { defineRpc, settingsRpc } from "@getpaseo/plugin";
import { DaemonClient } from "../test-utils/daemon-client.js";
import { createTestPaseoDaemon } from "../test-utils/paseo-daemon.js";
test("server reads saved settings after daemon restart before any client connects", async () => {
const root = await mkdtemp(path.join(tmpdir(), "settings-restart-"));
onTestFinished(() => rm(root, { recursive: true, force: true }));
const directory = path.join(root, "plugin");
const startupReport = path.join(root, "startup.json");
await mkdir(path.join(directory, "server"), { recursive: true });
await writeFile(
path.join(directory, "paseo-plugin.json"),
JSON.stringify({
id: "settings-startup",
requirements: { paseo: `>=${resolveDaemonVersion(import.meta.url)}` },
}),
);
await writeFile(
path.join(directory, "server", "report.ts"),
`import { writeFile } from "node:fs/promises";
export async function reportStartup(settings) {
const state = await settings.read();
await writeFile(${JSON.stringify(startupReport)}, JSON.stringify(state));
}`,
);
await writeFile(
path.join(directory, "index.server.ts"),
`import { defineSettings } from "@getpaseo/plugin";
import { z } from "zod";
import { reportStartup } from "./server/report";
export default function(server) {
const settings = server.registerSettings(defineSettings({ id: "display", scope: "host", version: 1, schema: z.object({ enabled: z.boolean().default(true) }) }));
const startup = reportStartup(settings);
return async () => { await startup; };
}`,
);
const options = {
paseoHomeRoot: path.join(root, "daemon"),
staticDir: path.join(root, "static"),
cleanup: false,
pluginsEnabled: true,
plugins: {
"settings-startup": { source: "directory" as const, path: directory, enabled: true },
},
};
const first = await createTestPaseoDaemon(options);
onTestFinished(() => first.close());
const rpc = settingsRpc("display");
const readStartup = async () =>
rpc.read.output.parse(JSON.parse(await readFile(startupReport, "utf8")));
await expect
.poll(readStartup)
.toEqual({ status: "ready", revision: "missing", values: { enabled: true } });
const client = new DaemonClient({ url: `ws://127.0.0.1:${first.port}/ws`, appVersion: "0.8.0" });
onTestFinished(() => client.close());
await client.connect();
const saved = rpc.write.output.parse(
await client.invokePluginRpc("settings-startup", rpc.write.name, {
revision: "missing",
values: { enabled: false },
}),
);
expect(saved).toMatchObject({ status: "saved", values: { enabled: false } });
const persisted = rpc.read.output.parse(
await client.invokePluginRpc("settings-startup", rpc.read.name, {}),
);
await client.close();
await first.close();
await rm(startupReport);
const restarted = await createTestPaseoDaemon(options);
onTestFinished(() => restarted.close());
// No client is created for this daemon: the report comes from server startup alone.
await expect.poll(readStartup).toEqual(persisted);
}, 60_000);
test("two clients share settings, observe changes, and preserve values through plugin lifecycle", async () => {
const directory = await mkdtemp(path.join(tmpdir(), "settings-plugin-"));
const daemon = await createTestPaseoDaemon();
const first = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws`, appVersion: "0.7.2" });
const second = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws`, appVersion: "0.7.2" });
const rpc = settingsRpc("display");
const serverReadRpc = defineRpc({
name: "settings-test.server-read",
input: z.object({}),
output: z.json(),
});
const serverChangesRpc = defineRpc({
name: "settings-test.server-changes",
input: z.object({}),
output: z.object({ count: z.number().int() }),
});
const read = async (client: DaemonClient, pluginId = "settings-test") =>
rpc.read.output.parse(await client.invokePluginRpc(pluginId, rpc.read.name, {}));
const changed: string[] = [];
@@ -27,9 +110,18 @@ test("two clients share settings, observe changes, and preserve values through p
);
await writeFile(
path.join(directory, "index.server.ts"),
`import { defineSettings } from "@getpaseo/plugin";
`import { defineRpc, defineSettings } from "@getpaseo/plugin";
import { z } from "zod";
export default function(server) { server.registerSettings(defineSettings({ id: "display", scope: "host", version: 1, schema: z.object({ enabled: z.boolean().default(true) }) })); return () => {}; }`,
const serverRead = defineRpc({ name: "settings-test.server-read", input: z.object({}), output: z.json() });
const serverChanges = defineRpc({ name: "settings-test.server-changes", input: z.object({}), output: z.object({ count: z.number().int() }) });
export default function(server) {
const settings = server.registerSettings(defineSettings({ id: "display", scope: "host", version: 1, schema: z.object({ enabled: z.boolean().default(true) }) }));
let changes = 0;
settings.subscribe(() => { changes += 1; });
server.handle(serverRead, () => settings.read());
server.handle(serverChanges, () => ({ count: changes }));
return () => {};
}`,
);
await first.connect();
await second.connect();
@@ -42,6 +134,16 @@ export default function(server) { server.registerSettings(defineSettings({ id: "
await first.installDirectoryPlugin(directory);
const initial = await read(first);
expect(initial).toMatchObject({ status: "ready", values: { enabled: true } });
expect(
serverReadRpc.output.parse(
await first.invokePluginRpc("settings-test", serverReadRpc.name, {}),
),
).toMatchObject({ status: "ready", values: { enabled: true } });
expect(
serverChangesRpc.output.parse(
await first.invokePluginRpc("settings-test", serverChangesRpc.name, {}),
),
).toEqual({ count: 0 });
expect(
await second.invokePluginRpc("settings-test", rpc.write.name, {
revision: initial.revision,
@@ -50,6 +152,16 @@ export default function(server) { server.registerSettings(defineSettings({ id: "
).toMatchObject({ status: "saved" });
await expect.poll(() => changed).toEqual(["display"]);
expect(await read(first)).toMatchObject({ values: { enabled: false } });
expect(
serverReadRpc.output.parse(
await first.invokePluginRpc("settings-test", serverReadRpc.name, {}),
),
).toMatchObject({ status: "ready", values: { enabled: false } });
expect(
serverChangesRpc.output.parse(
await first.invokePluginRpc("settings-test", serverChangesRpc.name, {}),
),
).toEqual({ count: 1 });
expect(
await first.invokePluginRpc("settings-test", rpc.write.name, {
revision: initial.revision,
@@ -3,6 +3,7 @@ import { tmpdir } from "node:os";
import path from "node:path";
import { afterEach, expect, test } from "vitest";
import { defineSettings } from "@getpaseo/plugin";
import type { PluginSettingsState } from "@getpaseo/plugin/server";
import { z } from "zod";
import { PluginSettingsStore } from "./index.js";
@@ -19,6 +20,10 @@ const definition = defineSettings({
count: z.number().int().min(1).default(5),
}),
});
type DisplaySettingsState = PluginSettingsState<typeof definition.schema>;
function mutateNotification(state: DisplaySettingsState): void {
if (state.status === "ready") state.values.count = 99;
}
async function setup() {
const directory = await mkdtemp(path.join(tmpdir(), "plugin-settings-"));
roots.push(directory);
@@ -47,6 +52,47 @@ test("defaults, atomic saves, concurrent revisions, and restart persistence", as
expect(changes).toEqual(["display"]);
});
test("server settings read current values and notify after successful writes", async () => {
const { handlers } = await setup();
const notifications: unknown[] = [];
const unsubscribe = handlers.settings.subscribe((state) => {
notifications.push(state);
mutateNotification(state);
});
expect(await handlers.settings.read()).toEqual({
status: "ready",
revision: "missing",
values: { enabled: true, count: 5 },
});
const saved = await handlers.write.handle({
revision: "missing",
values: { enabled: false, count: 10 },
});
expect(saved).toMatchObject({ status: "saved", values: { enabled: false, count: 10 } });
const current = await handlers.settings.read();
expect(current).toMatchObject({
status: "ready",
values: { enabled: false, count: 10 },
});
expect(notifications).toEqual([
{
status: "ready",
revision: current.revision,
values: { enabled: false, count: 99 },
},
]);
expect(
await handlers.write.handle({ revision: current.revision, values: { count: -1 } }),
).toMatchObject({ status: "invalid" });
expect(notifications).toHaveLength(1);
unsubscribe();
await handlers.reset.handle({ revision: current.revision });
expect(notifications).toHaveLength(1);
});
test("validation failure preserves disk and leaves the writer usable", async () => {
const { handlers, changes } = await setup();
expect(await handlers.write.handle({ revision: "missing", values: { count: -1 } })).toMatchObject(
@@ -1,9 +1,9 @@
import { createHash, randomUUID } from "node:crypto";
import { mkdir, readFile, rename, rm, writeFile } from "node:fs/promises";
import path from "node:path";
import { z } from "zod";
import { defineSettings, type SettingsDefinition } from "@getpaseo/plugin";
import { settingsRpc } from "@getpaseo/plugin";
import { z, type ZodType } from "zod";
import { defineSettings, settingsRpc, type SettingsDefinition } from "@getpaseo/plugin";
import type { PluginSettings, PluginSettingsState } from "@getpaseo/plugin/server";
const envelopeSchema = z.object({ version: z.number().int().positive(), values: z.json() });
function message(error: unknown): string {
@@ -12,6 +12,25 @@ function message(error: unknown): string {
}
const revisionOf = (raw: string) => createHash("sha256").update(raw).digest("hex");
type SettingsListener<Schema extends ZodType> = (
state: PluginSettingsState<Schema>,
) => void | Promise<void>;
type SettingsWriteState<Schema extends ZodType> =
| { status: "saved"; revision: string; values: z.output<Schema> }
| { status: "conflict"; error: string }
| { status: "invalid"; error: string };
function reportListenerError(id: string, error: unknown): void {
console.error(`Plugin settings subscriber failed for ${id}`, error);
}
function copySettingsState<Schema extends ZodType>(
state: PluginSettingsState<Schema>,
): PluginSettingsState<Schema> {
if (state.status === "invalid") return { ...state };
return { ...state, values: structuredClone(state.values) };
}
/** One instance per installation. The subprocess lifetime gives writes a single owner. */
export class PluginSettingsStore {
private readonly definitions = new Map<string, SettingsDefinition>();
@@ -24,24 +43,43 @@ export class PluginSettingsStore {
this.changed = changed;
}
register(definition: SettingsDefinition) {
register<Schema extends ZodType>(definition: SettingsDefinition<Schema>) {
defineSettings(definition);
if (this.definitions.has(definition.id))
throw new Error(`Duplicate settings: ${definition.id}`);
this.definitions.set(definition.id, definition);
const listeners = new Set<SettingsListener<Schema>>();
const notify = (state: PluginSettingsState<Schema>): void => {
for (const listener of listeners) {
try {
void Promise.resolve(listener(copySettingsState(state))).catch((error) =>
reportListenerError(definition.id, error),
);
} catch (error) {
reportListenerError(definition.id, error);
}
}
};
const rpc = settingsRpc(definition.id);
const read = () => this.serial(() => this.read(definition, notify));
const write = (input: z.output<typeof rpc.write.input>) =>
this.serial(() => this.write(definition, input.revision, input.values, "save", notify));
const reset = (input: z.output<typeof rpc.reset.input>) =>
this.serial(() => this.write(definition, input.revision, {}, "reset", notify));
const settings: PluginSettings<Schema> = {
read,
subscribe(listener) {
listeners.add(listener);
return () => {
listeners.delete(listener);
};
},
};
return {
read: { contract: rpc.read, handle: () => this.serial(() => this.read(definition)) },
write: {
contract: rpc.write,
handle: (input: z.output<typeof rpc.write.input>) =>
this.serial(() => this.write(definition, input.revision, input.values, "save")),
},
reset: {
contract: rpc.reset,
handle: (input: z.output<typeof rpc.reset.input>) =>
this.serial(() => this.write(definition, input.revision, {}, "reset")),
},
settings,
read: { contract: rpc.read, handle: read },
write: { contract: rpc.write, handle: write },
reset: { contract: rpc.reset, handle: reset },
};
}
@@ -62,9 +100,10 @@ export class PluginSettingsStore {
}
}
private async read(
definition: SettingsDefinition,
): Promise<z.output<ReturnType<typeof settingsRpc>["read"]["output"]>> {
private async read<Schema extends ZodType>(
definition: SettingsDefinition<Schema>,
notify: (state: PluginSettingsState<Schema>) => void,
): Promise<PluginSettingsState<Schema>> {
const stored = await this.stored(definition.id);
try {
const envelope = stored.raw === null ? null : envelopeSchema.parse(JSON.parse(stored.raw));
@@ -76,58 +115,71 @@ export class PluginSettingsStore {
throw new Error(`Settings version ${envelope.version} requires a migration`);
values = await definition.migrate(values, envelope.version);
}
const parsed = z.json().parse(await definition.schema.parseAsync(values));
const revision =
envelope && envelope.version !== definition.version
? await this.persist(definition, parsed)
: stored.revision;
return { status: "ready", values: parsed, revision };
const parsed = await definition.schema.parseAsync(values);
z.json().parse(parsed);
const migrated = envelope !== null && envelope.version !== definition.version;
const revision = migrated ? await this.persist(definition, parsed) : stored.revision;
const result: PluginSettingsState<Schema> = {
status: "ready",
values: parsed,
revision,
};
if (migrated) {
notify(result);
this.changed(definition.id);
}
return result;
} catch (error) {
return { status: "invalid", revision: stored.revision, error: message(error) };
}
}
private async write(
definition: SettingsDefinition,
private async write<Schema extends ZodType>(
definition: SettingsDefinition<Schema>,
revision: string,
values: unknown,
intent: "save" | "reset",
): Promise<z.output<ReturnType<typeof settingsRpc>["write"]["output"]>> {
notify: (state: PluginSettingsState<Schema>) => void,
): Promise<SettingsWriteState<Schema>> {
const stored = await this.stored(definition.id);
if (stored.revision !== revision)
return {
status: "conflict",
error: "Settings changed on another client. Reload before saving again.",
};
let parsed: z.output<ReturnType<typeof z.json>>;
let parsed: z.output<Schema>;
try {
if (intent === "save" && stored.raw !== null) {
const envelope = envelopeSchema.parse(JSON.parse(stored.raw));
if (envelope.version !== definition.version)
throw new Error("Reload or reset settings before saving a different schema version");
}
parsed = z.json().parse(await definition.schema.parseAsync(values));
parsed = await definition.schema.parseAsync(values);
z.json().parse(parsed);
} catch (error) {
return { status: "invalid", error: message(error) };
}
return { status: "saved", values: parsed, revision: await this.persist(definition, parsed) };
const nextRevision = await this.persist(definition, parsed);
notify({ status: "ready", values: parsed, revision: nextRevision });
this.changed(definition.id);
return { status: "saved", values: parsed, revision: nextRevision };
}
private async persist(
definition: SettingsDefinition,
values: z.output<ReturnType<typeof z.json>>,
private async persist<Schema extends ZodType>(
definition: SettingsDefinition<Schema>,
values: z.output<Schema>,
) {
const jsonValues = z.json().parse(values);
await mkdir(this.directory, { recursive: true });
const target = path.join(this.directory, `${definition.id}.json`);
const temporary = `${target}.${randomUUID()}.tmp`;
const raw = JSON.stringify({ version: definition.version, values });
const raw = JSON.stringify({ version: definition.version, values: jsonValues });
try {
await writeFile(temporary, raw, { mode: 0o600 });
await rename(temporary, target);
} finally {
await rm(temporary, { force: true });
}
this.changed(definition.id);
return revisionOf(raw);
}
}
+5
View File
@@ -117,6 +117,11 @@ later through `session.config`.
prompt, MCP servers, tool policy, model, mode, settings, opaque provider options, and persistence
preference.
For persisted host-scoped plugin settings, use the handle returned by `server.registerSettings()`.
Read it in catalog and session callbacks, and subscribe when the provider keeps in-memory
configuration. See [settings](./reference.md#persisted-values) for the `read()` and `subscribe()`
contract.
Create the native session with the selected catalog values, then publish its effective state:
```ts
+29 -3
View File
@@ -1044,9 +1044,22 @@ export const preferences = defineSettings({
});
```
Register it with `server.registerSettings(preferences)` in `index.server.ts` before returning
cleanup. This server entry is required for built-in persistence; a screen using its own data
can remain client-only.
Register it in `index.server.ts` before returning cleanup. The returned handle lets server code
read the document and react to changes. A screen using its own data can remain client-only.
```ts
export default function contribute(server: PluginServerContext) {
const settings = server.registerSettings(preferences);
settings.subscribe((next) => {
if (next.status === "ready") {
console.log("Settings changed", next.revision);
}
});
return () => {};
}
```
| Definition field | Contract |
| ------------------------------ | ------------------------------------------------------------------------------------------------------------------------------------ |
@@ -1065,6 +1078,19 @@ Call `useSettings(preferences)` in any contributed component. It returns a discr
| `invalid` | `error` and `revision`; stored data is preserved. |
| `error` | `error` from the read/connection. |
The server handle exposes `read()` and `subscribe()`. `read()` returns the same `ready` or
`invalid` state as the client hook, including the opaque revision. `subscribe()` returns a cleanup
function and receives a new `ready` state after a successful save, reset, or migration. Invalid
writes and revision conflicts do not notify subscribers. Listener failures are logged without
turning a committed write into a failed save.
```ts
const current = await settings.read();
if (current.status === "ready") {
// Use current.values and current.revision.
}
```
Every state also exposes `saving`, `saveError`, and these actions:
- `save(values, revision): Promise<boolean>` validates and saves a complete document. Returns