diff --git a/packages/app/src/plugins/subscriptions.test.tsx b/packages/app/src/plugins/subscriptions.test.tsx index c84a2790ed..971e841fbf 100644 --- a/packages/app/src/plugins/subscriptions.test.tsx +++ b/packages/app/src/plugins/subscriptions.test.tsx @@ -227,6 +227,16 @@ it("useSubscription reports a refusal, a failure after going live, and a clean c await act(async () => subscriptions[2].acknowledge()); act(() => subscriptions[2].closed()); expect(finished.result.current).toEqual({ status: "closed" }); + + // A handler that returns at once closes in the same tick the host accepts it. + const immediate = renderHook(() => useSubscription(changes, { ref: "d" }, () => undefined), { + wrapper: wrapper(plugin), + }); + await act(async () => { + subscriptions[3].acknowledge(); + subscriptions[3].closed(); + }); + expect(immediate.result.current).toEqual({ status: "closed" }); }); it("an event that fails the contract closes the subscription with an error", async () => { diff --git a/packages/client/src/connection/owned.test.ts b/packages/client/src/connection/owned.test.ts index eca52f1749..7e5c2e7202 100644 --- a/packages/client/src/connection/owned.test.ts +++ b/packages/client/src/connection/owned.test.ts @@ -83,6 +83,26 @@ test("cancellation before the ID waits for bootstrap and releases exactly that I expect(c.released).toEqual(["cancelled"]); }); +test("a release right after the host accepts still reports the subscription as accepted", async () => { + const c = connection(); + const handle = c.observe(); + c.requests[0]!("closing"); + const released = handle.release(); + await expect(handle.ready).resolves.toMatchObject({ subscriptionId: "closing" }); + await released; + expect(c.released).toEqual(["closing"]); +}); + +test("a release before the host accepts rejects ready", async () => { + const c = connection(); + const handle = c.observe(); + const released = handle.release(); + await expect(handle.ready).rejects.toThrow("Subscription released"); + c.requests[0]!("late"); + await released; + expect(c.released).toEqual(["late"]); +}); + test("surviving handles use new remote IDs after reconnect and released handles stay gone", async () => { const c = connection(); const a = c.observe(); diff --git a/packages/client/src/connection/owned.ts b/packages/client/src/connection/owned.ts index e165c7e588..1c6290be7d 100644 --- a/packages/client/src/connection/owned.ts +++ b/packages/client/src/connection/owned.ts @@ -154,7 +154,9 @@ export class OwnedSubscriptions { observers.clear(); buffered.length = 0; if (id) this.routes.delete(id); - reject(new Error("Subscription released")); + // A host that already accepted keeps `ready` resolved, even if released in the same tick. + if (snapshot) resolve(snapshot); + else reject(new Error("Subscription released")); teardown = (async () => { await pending; await releaseId(); diff --git a/packages/protocol/src/binary-frames/plugin.test.ts b/packages/protocol/src/binary-frames/plugin.test.ts index e3e53133e3..644e6235ef 100644 --- a/packages/protocol/src/binary-frames/plugin.test.ts +++ b/packages/protocol/src/binary-frames/plugin.test.ts @@ -8,6 +8,8 @@ import { encodePluginInputFrame, encodePluginOutputFrame, FileTransferOpcode, + MAX_PLUGIN_PAYLOAD_BYTES, + MAX_ROUTED_PAYLOAD_BYTES, PluginFrameOpcode, TerminalStreamOpcode, } from "./index.js"; @@ -70,6 +72,21 @@ describe("the plugin byte rule", () => { expect(() => assertPluginWireValue(new Uint8Array(4), "event")).not.toThrow(); }); + it("rejects bytes that would not fit in one frame with the largest header", () => { + const bytes = new Uint8Array(MAX_PLUGIN_PAYLOAD_BYTES + 1); + expect(() => assertPluginWireValue(bytes.subarray(1), "event")).not.toThrow(); + expect(() => assertPluginWireValue(bytes, "event")).toThrow( + `event is ${MAX_PLUGIN_PAYLOAD_BYTES + 1} bytes; one value carries at most ${MAX_PLUGIN_PAYLOAD_BYTES} bytes`, + ); + const frame = encodePluginInputFrame({ + requestId: "r".repeat(255), + pluginId: "p", + method: "m".repeat(0xffff - 30), + payload: bytes.subarray(1), + }); + expect(frame.byteLength).toBeLessThanOrEqual(MAX_ROUTED_PAYLOAD_BYTES); + }); + it("rejects bytes nested in a value and names the rule and the path", () => { expect(() => assertPluginWireValue({ frames: [{ data: new Uint8Array(1) }] }, "event")).toThrow( "event.frames[0].data is a Uint8Array. Bytes cross the wire only as the whole value: send the Uint8Array itself, or encode it into the value.", diff --git a/packages/protocol/src/binary-frames/plugin.ts b/packages/protocol/src/binary-frames/plugin.ts index 2177dbc192..a2db0addf9 100644 --- a/packages/protocol/src/binary-frames/plugin.ts +++ b/packages/protocol/src/binary-frames/plugin.ts @@ -1,3 +1,5 @@ +import { MAX_ROUTED_PAYLOAD_BYTES } from "./host-route.js"; + /** * Plugin values that are a `Uint8Array` travel as binary frames. Opcodes 0x20-0x2f are * reserved for plugin traffic. Every frame is `[opcode][id length u8][id]...` and the payload @@ -14,6 +16,12 @@ export const PluginFrameOpcode = { export type PluginFrameOpcode = (typeof PluginFrameOpcode)[keyof typeof PluginFrameOpcode]; +/** The largest frame header: an input's opcode, ID and header JSON, each at its maximum length. */ +const MAX_PLUGIN_FRAME_HEADER_BYTES = 2 + 0xff + 2 + 0xffff; + +/** The largest `Uint8Array` a plugin value carries, so its frame fits in one routed message. */ +export const MAX_PLUGIN_PAYLOAD_BYTES = MAX_ROUTED_PAYLOAD_BYTES - MAX_PLUGIN_FRAME_HEADER_BYTES; + export type PluginFrame = | { opcode: typeof PluginFrameOpcode.Event; subscriptionId: string; payload: Uint8Array } | { opcode: typeof PluginFrameOpcode.Output; requestId: string; payload: Uint8Array } @@ -112,7 +120,13 @@ function decodeInputHeader(bytes: Uint8Array): { pluginId: string; method: strin * with no `Uint8Array` inside it. Checks object structure only; never reads the bytes. */ export function assertPluginWireValue(value: unknown, label: string): void { - if (value instanceof Uint8Array) return; + if (value instanceof Uint8Array) { + if (value.byteLength > MAX_PLUGIN_PAYLOAD_BYTES) + throw new RangeError( + `${label} is ${value.byteLength} bytes; one value carries at most ${MAX_PLUGIN_PAYLOAD_BYTES} bytes`, + ); + return; + } const nested = findNestedBytes(value, label); if (nested) throw new TypeError( diff --git a/packages/server/src/server/plugins/runtime.posix.test.ts b/packages/server/src/server/plugins/runtime.posix.test.ts index cfa3292567..01a7cf5217 100644 --- a/packages/server/src/server/plugins/runtime.posix.test.ts +++ b/packages/server/src/server/plugins/runtime.posix.test.ts @@ -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 { diff --git a/packages/server/src/server/plugins/runtime.ts b/packages/server/src/server/plugins/runtime.ts index bd18662f47..98cbf8fcf0 100644 --- a/packages/server/src/server/plugins/runtime.ts +++ b/packages/server/src/server/plugins/runtime.ts @@ -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, diff --git a/packages/server/src/server/plugins/subscription-handlers.ts b/packages/server/src/server/plugins/subscription-handlers.ts index 79b993ec31..ec6e7baaec 100644 --- a/packages/server/src/server/plugins/subscription-handlers.ts +++ b/packages/server/src/server/plugins/subscription-handlers.ts @@ -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; diff --git a/packages/server/src/server/session.test.ts b/packages/server/src/server/session.test.ts index d1408f9880..acd292e895 100644 --- a/packages/server/src/server/session.test.ts +++ b/packages/server/src/server/session.test.ts @@ -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; + 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 = { diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 5f66fe9b58..c6cae8b74c 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -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[] { diff --git a/public-docs/plugins/reference.md b/public-docs/plugins/reference.md index 8c641a8d8f..0f5b3eb40a 100644 --- a/public-docs/plugins/reference.md +++ b/public-docs/plugins/reference.md @@ -2333,10 +2333,10 @@ union and branch with `event instanceof Uint8Array`. A `Uint8Array` nested in an validation on the sending side with an error that names the rule; send the bytes as their own event or RPC value and the metadata as a separate JSON event. -One event or RPC value carries at most 100 MiB. Push large payloads in chunks of a few hundred -KiB; each `push` waits for the previous one to leave the plugin process. Uploads from a phone have -a lower limit: Android closes the connection when one send exceeds 16 MiB, so split a larger -upload into several RPCs. +One event or RPC value carries at most 100 MiB less 65 KiB of framing. Push large payloads in +chunks of a few hundred KiB; each `push` waits for the previous one to leave the plugin process. +Uploads from a phone have a lower limit: Android closes the connection when one send exceeds +16 MiB, so split a larger upload into several RPCs. ### Several streams on one subscription diff --git a/public-docs/plugins/reference.zh-CN.md b/public-docs/plugins/reference.zh-CN.md index a07a94e8a9..7862ff781f 100644 --- a/public-docs/plugins/reference.zh-CN.md +++ b/public-docs/plugins/reference.zh-CN.md @@ -2016,7 +2016,7 @@ const png = await renderThumbnail({ path, width }); // Uint8Array 用 `z.instanceof(Uint8Array)` 声明字节。事件可以在每次推送时选择字节或 JSON:声明一个联合类型,并用 `event instanceof Uint8Array` 分支。嵌套在对象中的 `Uint8Array` 会在发送端校验失败,错误信息会说明这条规则;把字节作为单独的事件或 RPC 值发送,元数据作为另一个 JSON 事件发送。 -一个事件或 RPC 值最多 100 MiB。大载荷请按几百 KiB 分块推送;每次 `push` 都会等待上一次离开插件进程。手机上传的上限更低:一次发送超过 16 MiB 时 Android 会关闭连接,所以较大的上传请拆成多个 RPC。 +一个事件或 RPC 值最多 100 MiB(减去 65 KiB 的帧头)。大载荷请按几百 KiB 分块推送;每次 `push` 都会等待上一次离开插件进程。手机上传的上限更低:一次发送超过 16 MiB 时 Android 会关闭连接,所以较大的上传请拆成多个 RPC。 ### 在一个订阅中传输多条流 {#several-streams-on-one-subscription}