mirror of
https://github.com/getpaseo/paseo.git
synced 2026-11-03 07:03:18 +00:00
Mohamed Boudra
c689b5626e
Plugins can declare a subscription with defineSubscription, handle it with
server.handle(contract, (input, { push, signal })), and listen from a surface
with useSubscription or client.subscribe. A top-level Uint8Array in an event,
an RPC output, or an RPC input crosses the wire as a binary frame (opcodes
0x20-0x2f) without JSON or base64; a nested one fails validation on the sender.
Each subscriber runs its own handler, owned by the session that subscribed,
so release, disconnect, plugin stop and reload need no plugin code. The
runtime buffers pushes until the subscribe response has gone out, so a client
never sees an event for a subscription it does not know yet.
Gated on server_info.features.pluginSubscriptions; binary outputs reach only
clients that advertise plugin_binary_payloads.
415 lines
12 KiB
TypeScript
415 lines
12 KiB
TypeScript
import { z } from "zod";
|
|
import { describe, expect, it } from "vitest";
|
|
import {
|
|
MutableDaemonConfigSchema,
|
|
SessionInboundMessageSchema,
|
|
SessionOutboundMessageSchema,
|
|
StatusMessageSchema,
|
|
} from "./messages.js";
|
|
|
|
describe("plugin protocol compatibility", () => {
|
|
it.each([undefined, { paseo: ">=0.8.0" }])(
|
|
"parses plugin catalogs with requirements %j",
|
|
(requirements) => {
|
|
const message = {
|
|
type: "plugin.catalog.get.response",
|
|
payload: {
|
|
requestId: "catalog",
|
|
plugins: [
|
|
{ id: "example", clientBundle: "bundle", ...(requirements ? { requirements } : {}) },
|
|
],
|
|
},
|
|
};
|
|
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",
|
|
payload: {
|
|
requestId: "metadata",
|
|
plugins: [
|
|
{
|
|
id: "review",
|
|
name: "Review tools",
|
|
icon: "assets/icon.png",
|
|
media: ["screenshot.png", "https://example.com/demo.mp4"],
|
|
path: "/plugins/review",
|
|
enabled: false,
|
|
status: "disabled",
|
|
},
|
|
],
|
|
},
|
|
};
|
|
expect(SessionOutboundMessageSchema.parse(message)).toEqual(message);
|
|
});
|
|
|
|
it("parses plugin timeline append messages and advertises the capability", () => {
|
|
expect(
|
|
SessionInboundMessageSchema.parse({
|
|
type: "agent.timeline.append.request",
|
|
requestId: "request-append",
|
|
agentId: "agent-1",
|
|
item: {
|
|
type: "plugin",
|
|
id: "review-1",
|
|
kind: "review",
|
|
version: 1,
|
|
data: { status: "running" },
|
|
},
|
|
}),
|
|
).toMatchObject({ type: "agent.timeline.append.request" });
|
|
expect(
|
|
SessionOutboundMessageSchema.parse({
|
|
type: "agent.timeline.append.response",
|
|
payload: { requestId: "request-append", seq: 7, epoch: "epoch-1" },
|
|
}),
|
|
).toMatchObject({ type: "agent.timeline.append.response" });
|
|
expect(
|
|
StatusMessageSchema.parse({
|
|
type: "status",
|
|
payload: {
|
|
status: "server_info",
|
|
serverId: "server-1",
|
|
features: { pluginTimelineItems: true },
|
|
},
|
|
}).payload,
|
|
).toMatchObject({ features: { pluginTimelineItems: true } });
|
|
});
|
|
|
|
it.each([0, -1, 1.5])("rejects plugin timeline version %s", (version) => {
|
|
expect(() =>
|
|
SessionInboundMessageSchema.parse({
|
|
type: "agent.timeline.append.request",
|
|
requestId: "append-1",
|
|
agentId: "agent-1",
|
|
item: { type: "plugin", id: "row-1", kind: "review", version, data: {} },
|
|
}),
|
|
).toThrow();
|
|
});
|
|
|
|
it("keeps old directory plugin config valid when enabled is absent", () => {
|
|
const config = MutableDaemonConfigSchema.parse({
|
|
mcp: { injectIntoAgents: true },
|
|
plugins: { example: { source: "directory", path: "/plugins/example" } },
|
|
});
|
|
|
|
expect(config.plugins?.example?.enabled).toBeUndefined();
|
|
});
|
|
|
|
it("uses namespaced management request and response pairs", () => {
|
|
expect(
|
|
SessionInboundMessageSchema.parse({
|
|
type: "plugin.directory.install.request",
|
|
requestId: "request-1",
|
|
path: "/plugins/example",
|
|
id: "example-work",
|
|
}).type,
|
|
).toBe("plugin.directory.install.request");
|
|
expect(
|
|
SessionInboundMessageSchema.parse({
|
|
type: "plugin.directory.inspect.request",
|
|
requestId: "request-0",
|
|
path: "/plugins/example",
|
|
}).type,
|
|
).toBe("plugin.directory.inspect.request");
|
|
expect(
|
|
SessionOutboundMessageSchema.parse({
|
|
type: "plugin.directory.install.response",
|
|
payload: {
|
|
requestId: "request-1",
|
|
plugin: {
|
|
id: "example-work",
|
|
description: "Reviews changes before merge",
|
|
path: "/plugins/example",
|
|
enabled: true,
|
|
status: "running",
|
|
},
|
|
},
|
|
}).type,
|
|
).toBe("plugin.directory.install.response");
|
|
|
|
expect(
|
|
SessionOutboundMessageSchema.parse({
|
|
type: "plugin.directory.install.response",
|
|
payload: {
|
|
requestId: "request-old-shape",
|
|
plugin: {
|
|
id: "example-work",
|
|
path: "/plugins/example",
|
|
enabled: true,
|
|
status: "running",
|
|
},
|
|
},
|
|
}).type,
|
|
).toBe("plugin.directory.install.response");
|
|
});
|
|
|
|
it("uses capability-gated source management RPCs", () => {
|
|
expect(
|
|
SessionInboundMessageSchema.parse({
|
|
type: "plugin.source.install.request",
|
|
requestId: "request-install",
|
|
source: "owner/repository:plugins/review",
|
|
ref: "main",
|
|
}).type,
|
|
).toBe("plugin.source.install.request");
|
|
expect(
|
|
SessionInboundMessageSchema.parse({
|
|
type: "plugin.source.install.request",
|
|
requestId: "request-install-old-client",
|
|
source: "owner/repository",
|
|
pluginPath: "plugins/review",
|
|
}).type,
|
|
).toBe("plugin.source.install.request");
|
|
expect(
|
|
SessionOutboundMessageSchema.parse({
|
|
type: "plugin.source.status.response",
|
|
payload: {
|
|
requestId: "request-status",
|
|
plugins: [
|
|
{
|
|
id: "review",
|
|
source: "git",
|
|
path: "/plugins/review",
|
|
currentCommit: "a".repeat(40),
|
|
latestCommit: "b".repeat(40),
|
|
commitsBehind: 2,
|
|
updateAvailable: true,
|
|
},
|
|
],
|
|
},
|
|
}).type,
|
|
).toBe("plugin.source.status.response");
|
|
|
|
const older = StatusMessageSchema.parse({
|
|
type: "status",
|
|
payload: {
|
|
status: "server_info",
|
|
serverId: "older-host",
|
|
features: { pluginManagement: true },
|
|
},
|
|
});
|
|
expect(older.payload.features?.pluginGitManagement).toBeUndefined();
|
|
});
|
|
|
|
it("uses a namespaced snapshot RPC for structured plugin logs", () => {
|
|
expect(
|
|
SessionInboundMessageSchema.parse({
|
|
type: "plugin.logs.get.request",
|
|
requestId: "request-logs",
|
|
pluginId: "example",
|
|
}),
|
|
).toEqual({
|
|
type: "plugin.logs.get.request",
|
|
requestId: "request-logs",
|
|
pluginId: "example",
|
|
});
|
|
expect(
|
|
SessionOutboundMessageSchema.parse({
|
|
type: "plugin.logs.get.response",
|
|
payload: {
|
|
requestId: "request-logs",
|
|
pluginId: "example",
|
|
entries: [
|
|
{
|
|
sequence: 7,
|
|
timestamp: "2026-08-16T12:00:00.000Z",
|
|
stream: "stderr",
|
|
message: "failed to connect",
|
|
},
|
|
],
|
|
},
|
|
}),
|
|
).toEqual({
|
|
type: "plugin.logs.get.response",
|
|
payload: {
|
|
requestId: "request-logs",
|
|
pluginId: "example",
|
|
entries: [
|
|
{
|
|
sequence: 7,
|
|
timestamp: "2026-08-16T12:00:00.000Z",
|
|
stream: "stderr",
|
|
message: "failed to connect",
|
|
},
|
|
],
|
|
},
|
|
});
|
|
});
|
|
|
|
it("keeps the plugin logs capability optional for older server info", () => {
|
|
const older = StatusMessageSchema.parse({
|
|
type: "status",
|
|
payload: {
|
|
status: "server_info",
|
|
serverId: "older-host",
|
|
features: { pluginManagement: true },
|
|
},
|
|
});
|
|
const current = StatusMessageSchema.parse({
|
|
type: "status",
|
|
payload: {
|
|
status: "server_info",
|
|
serverId: "current-host",
|
|
features: { pluginManagement: true, pluginLogs: true },
|
|
},
|
|
});
|
|
|
|
expect(older.payload.features?.pluginLogs).toBeUndefined();
|
|
expect(current.payload.features?.pluginLogs).toBe(true);
|
|
});
|
|
|
|
it("keeps the plugin themes capability optional for older server info", () => {
|
|
const older = StatusMessageSchema.parse({
|
|
type: "status",
|
|
payload: {
|
|
status: "server_info",
|
|
serverId: "older-host",
|
|
features: { plugins: true },
|
|
},
|
|
});
|
|
const current = StatusMessageSchema.parse({
|
|
type: "status",
|
|
payload: {
|
|
status: "server_info",
|
|
serverId: "current-host",
|
|
features: { plugins: true, pluginThemes: true },
|
|
},
|
|
});
|
|
|
|
expect(older.payload.features?.pluginThemes).toBeUndefined();
|
|
expect(current.payload.features?.pluginThemes).toBe(true);
|
|
});
|
|
|
|
it("keeps the catalog change notification safe for older clients", () => {
|
|
expect(
|
|
StatusMessageSchema.parse({
|
|
type: "status",
|
|
payload: { status: "plugin_catalog_changed", pluginId: "example" },
|
|
}),
|
|
).toEqual({
|
|
type: "status",
|
|
payload: { status: "plugin_catalog_changed", pluginId: "example" },
|
|
});
|
|
});
|
|
|
|
it("requires plugin action payloads and keeps remove empty", () => {
|
|
expect(() =>
|
|
SessionOutboundMessageSchema.parse({
|
|
type: "plugin.reload.response",
|
|
payload: { requestId: "request-1" },
|
|
}),
|
|
).toThrow();
|
|
expect(
|
|
SessionOutboundMessageSchema.parse({
|
|
type: "plugin.remove.response",
|
|
payload: { requestId: "request-2" },
|
|
}),
|
|
).toEqual({ type: "plugin.remove.response", payload: { requestId: "request-2" } });
|
|
expect(() =>
|
|
SessionOutboundMessageSchema.parse({
|
|
type: "plugin.remove.response",
|
|
payload: { requestId: "request-2", plugin: { id: "extra" } },
|
|
}),
|
|
).toThrow();
|
|
});
|
|
});
|
|
|
|
it("npm metadata leaves legacy plugin rows parseable without extending their closed source enum", () => {
|
|
const legacyPlugin = z.object({
|
|
id: z.string(),
|
|
path: z.string(),
|
|
enabled: z.boolean(),
|
|
status: z.enum(["running", "disabled", "failed"]),
|
|
source: z.enum(["directory", "git"]).optional(),
|
|
});
|
|
const plugin = {
|
|
id: "review",
|
|
path: "/plugins/review",
|
|
enabled: true,
|
|
status: "running",
|
|
installation: {
|
|
identity: { kind: "npm", packageName: "@acme/review", pluginPath: "." },
|
|
currentRevision: "1.2.0",
|
|
},
|
|
};
|
|
const message = SessionOutboundMessageSchema.parse({
|
|
type: "plugin.list.response",
|
|
payload: { requestId: "npm", plugins: [plugin] },
|
|
});
|
|
expect(legacyPlugin.parse(plugin)).toEqual({
|
|
id: "review",
|
|
path: "/plugins/review",
|
|
enabled: true,
|
|
status: "running",
|
|
});
|
|
expect(message).toEqual({
|
|
type: "plugin.list.response",
|
|
payload: { requestId: "npm", plugins: [plugin] },
|
|
});
|
|
});
|
|
|
|
describe("plugin subscription protocol", () => {
|
|
it("parses a subscribe request and its response", () => {
|
|
const request = {
|
|
type: "plugin.rpc.subscribe.request",
|
|
requestId: "r1",
|
|
pluginId: "diff-watch",
|
|
method: "diff.watch",
|
|
input: { a: "main", b: "HEAD" },
|
|
};
|
|
expect(SessionInboundMessageSchema.parse(request)).toEqual(request);
|
|
const response = {
|
|
type: "plugin.rpc.subscribe.response",
|
|
payload: { requestId: "r1", subscriptionId: "s1" },
|
|
};
|
|
expect(SessionOutboundMessageSchema.parse(response)).toEqual(response);
|
|
});
|
|
|
|
it("parses JSON events and closes with and without an error", () => {
|
|
for (const message of [
|
|
{ type: "plugin.rpc.event", payload: { subscriptionId: "s1", event: { path: "a.ts" } } },
|
|
{ type: "plugin.rpc.closed", payload: { subscriptionId: "s1" } },
|
|
{ type: "plugin.rpc.closed", payload: { subscriptionId: "s1", error: "boom" } },
|
|
]) {
|
|
expect(SessionOutboundMessageSchema.parse(message)).toEqual(message);
|
|
}
|
|
});
|
|
|
|
it("advertises the feature as an optional flag", () => {
|
|
const status = {
|
|
type: "status",
|
|
payload: {
|
|
status: "server_info",
|
|
serverId: "srv",
|
|
features: { pluginSubscriptions: true },
|
|
},
|
|
};
|
|
expect(StatusMessageSchema.parse(status)).toMatchObject(status);
|
|
});
|
|
});
|