fix(plugins): close subscription gaps found in review

- Losing daemon.manage now releases the session's plugin subscriptions;
  binary events bypass the outbound permission check, so streams would
  otherwise outlive the revocation.
- A subscribe that times out sends unsubscribe, since the plugin may
  still start the handler after the daemon gave up.
- The byte limit leaves room for the largest frame header and routing,
  and lives in assertPluginWireValue so events, outputs and inputs share it.
- A release right after the host accepted keeps `ready` resolved, so a
  handler that returns at once reports closed rather than an error.
- Session trace logging tolerates loggers without isLevelEnabled.
This commit is contained in:
Mohamed Boudra
2026-11-02 05:51:55 +01:00
parent c689b5626e
commit 9233e8cd38
12 changed files with 144 additions and 13 deletions
@@ -1399,6 +1399,42 @@ export default function contribute(server: { registerProvider(provider: Provider
}
});
it("releases a subscription in the plugin when its subscribe times out", async () => {
vi.useFakeTimers();
try {
const directory = await createPlugin(
"slow-subscribe",
`export default function contribute(plugin: unknown) { void plugin; return () => undefined; }`,
);
const child = createReloadChild("slow-subscribe", []);
const sent: Array<{ type: string; subscriptionId?: string }> = [];
const originalSend = child.send.bind(child);
child.send = (message, callback) => {
sent.push(message);
if (message.type !== "initialize") return originalSend(message, callback);
callback?.(null);
queueMicrotask(() =>
child.emitMessage({ type: "ready", methods: [], subscriptions: ["feed"], providers: [] }),
);
return true;
};
const runtime = createTestRuntime({ spawnChild: () => child });
await runtime.startPlugin("slow-subscribe", directory);
const subscription = runtime.startSubscription("slow-subscribe", "feed", {});
const rejection = subscription.ready.catch((error: Error) => error);
await vi.advanceTimersByTimeAsync(30_000);
await expect(rejection).resolves.toMatchObject({ message: expect.stringContaining("timed out") });
const subscribe = sent.find((message) => message.type === "subscribe");
expect(sent).toContainEqual({
type: "unsubscribe",
subscriptionId: subscribe?.subscriptionId,
});
} finally {
vi.useRealTimers();
}
});
it("escalates a plugin that ignores graceful shutdown from TERM to KILL", async () => {
vi.useFakeTimers();
try {
@@ -767,6 +767,11 @@ export class PluginRuntime {
if (message.type === "hook") {
void send(child, { type: "hook.cancel", requestId }).catch(() => {});
}
// The handler may still start after the daemon gave up on it.
if (message.type === "subscribe") {
const { subscriptionId } = message;
void send(child, { type: "unsubscribe", subscriptionId }).catch(() => {});
}
reject(new Error(`Plugin RPC timed out: ${pluginId}.${message.type}`));
},
message.type === "tool.call" ? TOOL_CALL_TIMEOUT_MS : REQUEST_TIMEOUT_MS,
@@ -2,7 +2,6 @@ import type { PluginSubscriptionContract } from "@getpaseo/plugin";
import type { PluginSubscriptionContext } from "@getpaseo/plugin/server";
import type { PaseoApi } from "@getpaseo/client";
import { assertPluginWireValue } from "@getpaseo/protocol/binary-frames/index";
import { MAX_WEBSOCKET_MESSAGE_BYTES } from "@getpaseo/protocol/websocket-message-limit";
import type { PluginProcessMessage, PluginProcessRequest } from "./plugin-process-protocol.js";
export type PluginSubscriptionHandler = (
@@ -80,10 +79,6 @@ export class PluginSubscriptionHandlers {
if (signal.aborted) return;
const value: unknown = registered.contract.event.parse(event);
assertPluginWireValue(value, "event");
if (value instanceof Uint8Array && value.byteLength > MAX_WEBSOCKET_MESSAGE_BYTES)
throw new RangeError(
`event is ${value.byteLength} bytes; one event carries at most ${MAX_WEBSOCKET_MESSAGE_BYTES} bytes`,
);
// Events pushed while the handler runs follow the acknowledgement.
await acknowledged;
if (signal.aborted) return;
@@ -601,6 +601,29 @@ test("routes plugin requests and releases its owned catalog subscription on clea
expect(releasePluginSubscription).toHaveBeenCalledOnce();
});
test("losing daemon.manage releases the session's plugin subscriptions", async () => {
const messages: SessionOutboundMessage[] = [];
const release = vi.fn();
const pluginRuntime = {
subscribe: () => () => {},
subscribePluginRpc: () => ({ ready: Promise.resolve(), attach: () => {}, release }),
} as unknown as NonNullable<SessionOptions["pluginRuntime"]>;
const session = createSessionForTest({ messages, pluginRuntime });
await session.handleMessage({
type: "plugin.rpc.subscribe.request",
requestId: "watch",
pluginId: "example",
method: "screen.capture",
input: {},
});
expect(messages.map((message) => message.type)).toContain("plugin.rpc.subscribe.response");
session.setPermissions(["daemon.read"]);
// Bytes have no outbound permission check, so the stream itself has to end.
await vi.waitFor(() => expect(release).toHaveBeenCalledOnce());
});
test("leaves out the client bundles a catalog request already holds", async () => {
const messages: SessionOutboundMessage[] = [];
const pluginRuntime: NonNullable<SessionOptions["pluginRuntime"]> = {
+10 -1
View File
@@ -2231,7 +2231,8 @@ export class Session {
this.peakInflightRequests = this.inflightRequests;
}
try {
if (this.sessionLogger.isLevelEnabled("trace"))
// Stringifying a binary plugin input costs its size; test logger stubs lack isLevelEnabled.
if (this.sessionLogger.isLevelEnabled?.("trace"))
this.sessionLogger.trace(
{
messageType: msg.type,
@@ -2303,6 +2304,14 @@ export class Session {
this.sessionLogger.error({ err: error }, "Failed to release revoked browser hosting"),
);
}
// Binary plugin events skip the outbound permission check, so revocation ends the streams.
if (!this.authorization.allowsPermission("daemon.manage")) {
void this.delivery
.releaseFamily("plugin-rpc")
.catch((error) =>
this.sessionLogger.error({ err: error }, "Failed to release revoked plugin subscriptions"),
);
}
}
public getPermissions(): DaemonPermission[] {