diff --git a/packages/server/package.json b/packages/server/package.json index 3f03281b72..e2a057485c 100644 --- a/packages/server/package.json +++ b/packages/server/package.json @@ -53,7 +53,7 @@ "speech:transcribe:local": "tsx scripts/transcribe-local-wav.ts", "test": "npm run test:unit && npm run test:integration", "test:unit": "vitest run --fileParallelism --exclude \"**/*.e2e.test.ts\"", - "test:integration": "vitest run --maxWorkers=1 src/server/daemon-e2e/terminal-workspace-sdk.e2e.test.ts src/server/daemon-e2e/models.e2e.test.ts src/server/daemon-e2e/live-preferences.e2e.test.ts src/server/agent/model-catalog.e2e.test.ts && vitest run --maxWorkers=1 src/server/session.create-agent-worktree-autoarchive.e2e.test.ts -t \"creates a worktree and auto-archives both\"", + "test:integration": "vitest run --maxWorkers=1 src/server/daemon-e2e/terminal-workspace-sdk.e2e.test.ts src/server/daemon-e2e/models.e2e.test.ts src/server/daemon-e2e/live-preferences.e2e.test.ts src/server/agent/model-catalog.e2e.test.ts src/server/plugins/plugin-paseo-api.e2e.test.ts && vitest run --maxWorkers=1 src/server/session.create-agent-worktree-autoarchive.e2e.test.ts -t \"creates a worktree and auto-archives both\"", "test:hub-cli-contract": "vitest run --maxWorkers=1 src/server/hub/hub-cli-contract.test.ts", "test:integration:all": "npm run test:e2e", "test:integration:real": "vitest run real.e2e.test.ts", diff --git a/packages/server/src/server/plugins/plugin-paseo-api.e2e.test.ts b/packages/server/src/server/plugins/plugin-paseo-api.e2e.test.ts index e2bd3f0566..d0ad016cec 100644 --- a/packages/server/src/server/plugins/plugin-paseo-api.e2e.test.ts +++ b/packages/server/src/server/plugins/plugin-paseo-api.e2e.test.ts @@ -1,8 +1,10 @@ +import { EventEmitter, once } from "node:events"; import { resolveDaemonVersion } from "../daemon-version.js"; import { mkdir, mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import path from "node:path"; import { afterEach, expect, test } from "vitest"; +import { z } from "zod"; import { DaemonClient } from "../test-utils/daemon-client.js"; import { createTestPaseoDaemon } from "../test-utils/paseo-daemon.js"; import { createTestAgentClient, createTestAgentClients } from "../test-utils/fake-agent-client.js"; @@ -178,6 +180,7 @@ test("daemon config reload enables and disables configured plugins without resta appVersion: "0.4.0", }); const configPath = path.join(daemon.paseoHome, "config.json"); + const catalogChanges = new EventEmitter(); async function setPluginsEnabled(enabled: boolean): Promise { const config = JSON.parse(await readFile(configPath, "utf8")); @@ -189,10 +192,25 @@ test("daemon config reload enables and disables configured plugins without resta try { await client.connect(); + const catalog = client.observeEvents(["status.plugin_catalog_changed"]); + catalog.subscribe({ + snapshot: () => undefined, + update: (message) => { + if ( + message.type === "status" && + message.payload.status === "plugin_catalog_changed" && + message.payload.pluginId === "reloadable-plugin" + ) { + catalogChanges.emit("changed"); + } + }, + }); + await catalog.ready; await expect(client.listPlugins()).resolves.toEqual([ expect.objectContaining({ id: "reloadable-plugin", status: "disabled" }), ]); + const enabled = once(catalogChanges, "changed"); await setPluginsEnabled(true); await expect(client.reloadDaemonConfig()).resolves.toMatchObject({ requestId: expect.any(String), @@ -200,9 +218,12 @@ test("daemon config reload enables and disables configured plugins without resta restartRequiredPaths: [], overrideControlledPaths: [], }); - await expect - .poll(async () => (await client.listPlugins()).find(({ id }) => id === "reloadable-plugin")) - .toMatchObject({ enabled: true, status: "running" }); + await enabled; + expect((await client.listPlugins()).find(({ id }) => id === "reloadable-plugin")).toMatchObject( + { enabled: true, status: "running" }, + ); + + const disabled = once(catalogChanges, "changed"); await setPluginsEnabled(false); await expect(client.reloadDaemonConfig()).resolves.toEqual({ @@ -211,11 +232,179 @@ test("daemon config reload enables and disables configured plugins without resta restartRequiredPaths: [], overrideControlledPaths: [], }); - await expect - .poll(async () => (await client.listPlugins()).find(({ id }) => id === "reloadable-plugin")) - .toMatchObject({ enabled: true, status: "disabled" }); + await disabled; + expect((await client.listPlugins()).find(({ id }) => id === "reloadable-plugin")).toMatchObject( + { enabled: true, status: "disabled" }, + ); + await catalog.release(); } finally { await client.close().catch(() => undefined); await daemon.close(); } }, 60_000); + +const ReconnectingPluginStateSchema = z.object({ + snapshots: z.array(z.string()), + names: z.array(z.string()), +}); + +async function readReconnectingPluginState(client: DaemonClient) { + return ReconnectingPluginStateSchema.parse( + await client.invokePluginRpc("reconnecting", "state", {}), + ); +} + +test("plugin host APIs and observations recover after repeated daemon-side socket closes", async () => { + const pluginDirectory = await mkdtemp(path.join(tmpdir(), "paseo-reconnecting-plugin-")); + const workspaceDirectory = await mkdtemp(path.join(tmpdir(), "paseo-reconnecting-workspace-")); + roots.push(pluginDirectory, workspaceDirectory); + await writeFile( + path.join(pluginDirectory, "paseo-plugin.json"), + JSON.stringify({ id: "reconnecting", requirements: { paseo: ">=0.8.0" } }), + ); + await writeFile( + path.join(pluginDirectory, "index.server.ts"), + ` +import { defineRpc } from "@getpaseo/plugin"; +import { type PluginServerContext } from "@getpaseo/plugin/server"; +import { z } from "zod"; + +export default function contribute(server: PluginServerContext) { + const snapshots: string[] = []; + const names: string[] = []; + const snapshotWaiters = new Map void>(); + const nameWaiters = new Map void>(); + let reconnected = Promise.resolve(); + let release: (() => Promise) | undefined; + server.handle(defineRpc({ name: "observe", input: z.object({}), output: z.string() }), async (_, { paseo }) => { + const { subscription } = await paseo.workspaces.list({ subscribe: {} }); + subscription.subscribe({ + snapshot: (snapshot) => { + const id = snapshot.subscriptionId!; + snapshots.push(id); + snapshotWaiters.get(snapshots.length)?.(id); + snapshotWaiters.delete(snapshots.length); + }, + update: (message) => { + if (message.type === "workspace_update" && message.payload.kind === "upsert") { + const name = message.payload.workspace.name; + names.push(name); + nameWaiters.get(name)?.(name); + nameWaiters.delete(name); + } + }, + }); + release = () => subscription.release(); + return subscription.subscriptionId!; + }); + server.handle(defineRpc({ name: "wait-snapshot", input: z.object({ count: z.number().int().positive() }), output: z.string() }), ({ count }) => { + const id = snapshots[count - 1]; + return id ?? new Promise((resolve) => snapshotWaiters.set(count, resolve)); + }); + server.handle(defineRpc({ name: "wait-name", input: z.object({ name: z.string() }), output: z.string() }), ({ name }) => { + return names.includes(name) ? name : new Promise((resolve) => nameWaiters.set(name, resolve)); + }); + server.handle(defineRpc({ name: "state", input: z.object({}), output: z.object({ snapshots: z.array(z.string()), names: z.array(z.string()) }) }), () => ({ snapshots, names })); + server.handle(defineRpc({ name: "probe", input: z.object({}), output: z.object({ pid: z.number(), projectIds: z.array(z.string()) }) }), async (_, { paseo }) => { + await reconnected; + return { pid: process.pid, projectIds: (await paseo.projects.list()).projects.map((project) => project.projectId) }; + }); + server.handle(defineRpc({ name: "release", input: z.object({}), output: z.null() }), async () => { + await release?.(); + return null; + }); + server.handle(defineRpc({ name: "disconnect", input: z.object({}), output: z.null() }), () => new Promise((resolve) => { + // Inject a protocol violation through real IPC. The real daemon closes the + // active socket, and the worker's normal transport must observe that close. + function closed(message: { type: string }) { + if (message.type !== "paseo_close") return; + process.off("message", closed); + resolve(null); + } + process.on("message", closed); + reconnected = new Promise((connected) => { + function ready(message: { type: string; data?: unknown }) { + if (message.type !== "paseo_frame" || typeof message.data !== "string") return; + const frame = JSON.parse(message.data); + if (frame.type !== "session" || frame.message.type !== "status" || frame.message.payload.status !== "server_info") return; + process.off("message", ready); + connected(); + } + process.on("message", ready); + }); + process.send!({ type: "paseo_frame", isBinary: false, data: JSON.stringify({ + type: "hello", clientId: "plugin:reconnecting", clientType: "cli", protocolVersion: 1, + }) }); + })); + return async () => { await release?.(); }; +}`, + ); + const daemon = await createTestPaseoDaemon(); + const client = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws` }); + let projectId: string | undefined; + try { + await client.connect(); + const opened = await client.openProject(workspaceDirectory); + if (!opened.workspace) throw new Error(opened.error ?? "Workspace did not open"); + projectId = opened.workspace.projectId; + const workspaceId = opened.workspace.id; + await client.patchDaemonConfig({ pluginsEnabled: true }); + await client.installDirectoryPlugin(pluginDirectory); + const original = await client.invokePluginRpc("reconnecting", "probe", {}); + expect(original).toEqual({ pid: expect.any(Number), projectIds: [projectId] }); + const initialId = await client.invokePluginRpc("reconnecting", "observe", {}); + const app = await client.fetchWorkspaces({ subscribe: {} }); + const appUpdates = new EventEmitter(); + app.subscription.subscribe({ + snapshot: () => undefined, + update: (message) => { + if (message.type === "workspace_update" && message.payload.kind === "upsert") { + appUpdates.emit(message.payload.workspace.name); + } + }, + }); + await client.setWorkspaceTitle(workspaceId, "Before disconnect"); + await expect( + client.invokePluginRpc("reconnecting", "wait-name", { name: "Before disconnect" }), + ).resolves.toBe("Before disconnect"); + const ids = [initialId]; + for (const name of ["First recovery", "Second recovery"]) { + await client.invokePluginRpc("reconnecting", "disconnect", {}); + const id = await client.invokePluginRpc("reconnecting", "wait-snapshot", { + count: ids.length + 1, + }); + ids.push(id); + expect((await readReconnectingPluginState(client)).snapshots).toEqual(ids); + expect(new Set(ids).size).toBe(ids.length); + expect(app.subscription.subscriptionId).toBe(app.subscriptionId); + const appUpdate = once(appUpdates, name); + await client.setWorkspaceTitle(workspaceId, name); + await expect(client.invokePluginRpc("reconnecting", "wait-name", { name })).resolves.toBe( + name, + ); + await appUpdate; + expect(await client.invokePluginRpc("reconnecting", "probe", {})).toEqual(original); + } + await client.invokePluginRpc("reconnecting", "release", {}); + const releasedState = await client.invokePluginRpc("reconnecting", "state", {}); + await client.invokePluginRpc("reconnecting", "disconnect", {}); + expect(await client.invokePluginRpc("reconnecting", "probe", {})).toEqual(original); + const appUpdateAfterRelease = once(appUpdates, "After release"); + await client.setWorkspaceTitle(workspaceId, "After release"); + await appUpdateAfterRelease; + await client.invokePluginRpc("reconnecting", "probe", {}); + expect(await client.invokePluginRpc("reconnecting", "state", {})).toEqual(releasedState); + await app.subscription.release(); + // Removal while reconnect is scheduled must stop the child, not reload it. + await client.invokePluginRpc("reconnecting", "disconnect", {}); + await client.removePlugin("reconnecting"); + expect(await client.listPlugins()).toEqual([]); + await expect(client.invokePluginRpc("reconnecting", "probe", {})).rejects.toThrow( + "Plugin is not available", + ); + } finally { + if (projectId) await client.removeProject(projectId); + await client.close(); + await daemon.close(); + } +}, 60_000); diff --git a/packages/server/src/server/plugins/plugin-process.ts b/packages/server/src/server/plugins/plugin-process.ts index 29523a7c00..41310f78e8 100644 --- a/packages/server/src/server/plugins/plugin-process.ts +++ b/packages/server/src/server/plugins/plugin-process.ts @@ -260,7 +260,8 @@ async function initialize(message: Extract(); + // One entry per hello handshake: an attached socket is not a redialled one. + const hellos: object[] = []; + const waiters = new Set<{ count: number; resolve: () => void }>(); + function recordHello(socket: object): void { + hellos.push(socket); + for (const waiter of waiters) { + if (hellos.length < waiter.count) continue; + waiters.delete(waiter); + waiter.resolve(); + } + } return { active, + hellos, + // Resolves on the handshake itself rather than after a delay, so the test + // waits for the event it cares about instead of a guess at how long it takes. + waitForHellos(count: number): Promise { + if (hellos.length >= count) return Promise.resolve(); + return new Promise((resolve) => { + waiters.add({ count, resolve }); + }); + }, host: { async attachPluginSocket(_pluginId: string, socket: PluginSessionSocket) { const closed = new Promise((resolve) => socket.once("close", resolve)); @@ -135,6 +155,7 @@ function createTrackedSessionHost() { if (typeof data !== "string") return; const message = JSON.parse(data); if (message.type !== "hello") return; + recordHello(socket); socket.send( JSON.stringify({ type: "session", @@ -1774,4 +1795,149 @@ export default function contribute(plugin: any) { "Plugin is not available", ); }); + + it("re-attaches a plugin session when the daemon closes the socket under a live child", async () => { + const directory = await createPlugin( + "reattaching", + `export default function contribute(plugin: unknown) { void plugin; return () => undefined; }`, + ); + const sessions = createTrackedSessionHost(); + const runtime = createTestRuntime({ sessionHost: sessions.host }); + await runtime.startPlugin("reattaching", directory); + + expect(sessions.active.size).toBe(1); + expect(sessions.hellos).toHaveLength(1); + const first = [...sessions.active][0] as PluginSessionSocket; + + // What an expired lease does: drop the socket, leave the process running. + first.close(1000, "expired application lease"); + + // A second handshake, not just a second attachment: the client really redialled. + await sessions.waitForHellos(2); + + expect(sessions.active.size).toBe(1); + expect([...sessions.active][0]).not.toBe(first); + await runtime.stopAll(); + expect(sessions.active.size).toBe(0); + }); + + it("replaces a closed plugin session only once the child sends a fresh hello", async () => { + const directory = await createPlugin( + "lazy-reattach", + `export default function contribute(plugin: unknown) { void plugin; return () => undefined; }`, + ); + const sessions = createTrackedSessionHost(); + const child = createReloadChild("lazy-reattach", []); + const runtime = createTestRuntime({ spawnChild: () => child, sessionHost: sessions.host }); + await runtime.startPlugin("lazy-reattach", directory); + const first = [...sessions.active][0] as PluginSessionSocket; + // This child is a stub with no real client, so no handshake has happened. + expect(sessions.hellos).toHaveLength(0); + + first.close(1000, "expired application lease"); + expect(sessions.active.size).toBe(0); + + // Frames the child had already queued belong to the session that just died. + // Standing a socket up for one would leave it unspoken to until the host's + // hello timeout closed it, and that close would stand up another. + child.emitMessage({ + type: "paseo_frame", + data: JSON.stringify({ type: "session", message: { type: "ping" } }), + isBinary: false, + }); + expect(sessions.active.size).toBe(0); + + child.emitMessage({ + type: "paseo_frame", + data: JSON.stringify({ + type: "hello", + clientId: "plugin:lazy-reattach", + clientType: "cli", + protocolVersion: 1, + }), + isBinary: false, + }); + await sessions.waitForHellos(1); + + expect(sessions.active.size).toBe(1); + expect([...sessions.active][0]).not.toBe(first); + await runtime.stopAll(); + expect(sessions.active.size).toBe(0); + }); + + it("attaches no replacement session when the plugin is stopped", async () => { + const directory = await createPlugin( + "stopping", + `export default function contribute(plugin: unknown) { void plugin; return () => undefined; }`, + ); + const sessions = createTrackedSessionHost(); + const runtime = createTestRuntime({ sessionHost: sessions.host }); + await runtime.startPlugin("stopping", directory); + expect(sessions.active.size).toBe(1); + + await runtime.stopPluginById("stopping"); + + expect(runtime.getLogs("stopping").map((entry) => entry.message)).not.toContain( + "[paseo] Re-attached plugin session", + ); + expect(sessions.active.size).toBe(0); + }); + it("closes an in-flight replacement when the plugin stops before attachment completes", async () => { + const directory = await createPlugin( + "stopping-redial", + `export default function contribute() { return () => undefined; }`, + ); + const child = createReloadChild("stopping-redial", []); + const sessions = createTrackedSessionHost(); + let finishAttachment!: () => void; + const attachmentHeld = new Promise((resolve) => { + finishAttachment = resolve; + }); + let replacementAttached!: (socket: PluginSessionSocket) => void; + const replacementStarted = new Promise((resolve) => { + replacementAttached = resolve; + }); + let attachments = 0; + const runtime = createTestRuntime({ + spawnChild: () => child, + sessionHost: { + async attachPluginSocket(pluginId, socket) { + const attachment = await sessions.host.attachPluginSocket(pluginId, socket); + attachments += 1; + if (attachments === 2) { + replacementAttached(socket); + await attachmentHeld; + } + return attachment; + }, + }, + }); + try { + await runtime.startPlugin("stopping-redial", directory); + const first = [...sessions.active][0] as PluginSessionSocket; + first.close(); + child.emitMessage({ + type: "paseo_frame", + data: JSON.stringify({ type: "hello" }), + isBinary: false, + }); + const replacement = await replacementStarted; + const replacementClosed = new Promise((resolve) => replacement.once("close", resolve)); + await runtime.stopPluginById("stopping-redial"); + finishAttachment(); + await replacementClosed; + expect(sessions.active.size).toBe(0); + expect(sessions.hellos).toEqual([]); + expect(runtime.catalog()).toEqual([]); + child.emitMessage({ + type: "paseo_frame", + data: JSON.stringify({ type: "hello" }), + isBinary: false, + }); + expect(attachments).toBe(2); + } finally { + finishAttachment(); + await runtime.stopAll(); + } + }); }); diff --git a/packages/server/src/server/plugins/runtime.ts b/packages/server/src/server/plugins/runtime.ts index 7affa3cd97..16d0fd0571 100644 --- a/packages/server/src/server/plugins/runtime.ts +++ b/packages/server/src/server/plugins/runtime.ts @@ -74,6 +74,22 @@ interface LoadedPlugin { sessionClosed: Promise | null; } +/** The session socket a loaded plugin currently speaks through. */ +interface PluginSessionBinding { + socket: PluginSessionSocket; + reattaching: boolean; + plugin: LoadedPlugin | null; +} + +interface PluginFrameInput { + session: PluginSessionBinding; + pluginId: string; + child: PluginChild; + sessionHost: PluginPaseoSessionHost; + frame: string | Uint8Array; + isBinary: boolean; +} + interface PendingProviderSend { input: ProviderInput; resolve: () => void; @@ -565,10 +581,14 @@ export class PluginRuntime { const outputCapture = new PluginOutputCapture(child, (stream, message) => { this.appendLog(pluginId, stream, message); }); - const sessionSocket = new PluginSessionSocket(child); + const session: PluginSessionBinding = { + socket: new PluginSessionSocket(child), + reattaching: false, + plugin: null, + }; const pending = new Map(); const sessionAttachment = await sessionHost - .attachPluginSocket(pluginId, sessionSocket) + .attachPluginSocket(pluginId, session.socket) .catch((error) => { terminatePluginChild(child); throw error; @@ -601,9 +621,16 @@ export class PluginRuntime { } const message = parsed.data; if (message.type === "paseo_frame") { - sessionSocket.receive(message.data, message.isBinary); + this.routePluginFrame({ + session, + pluginId, + child, + sessionHost, + frame: message.data, + isBinary: message.isBinary, + }); } else if (message.type === "paseo_close") { - sessionSocket.peerClosed(); + session.socket.peerClosed(); } else if (message.type === "ready") { if (settled) return; settled = true; @@ -616,7 +643,7 @@ export class PluginRuntime { } }); child.on("close", () => { - sessionSocket.peerClosed(); + session.socket.peerClosed(); if (!loaded) { fail(new Error(`Plugin ${pluginId} exited during initialization`)); return; @@ -635,7 +662,7 @@ export class PluginRuntime { }, ); } catch (error) { - sessionSocket.close(); + session.socket.close(); await sessionAttachment.closed; terminatePluginChild(child); throw error; @@ -652,9 +679,10 @@ export class PluginRuntime { pending, providerConnections: new Map(), providerConnectionTombstones: new Set(), - sessionSocket, + sessionSocket: session.socket, sessionClosed: sessionAttachment.closed, }; + session.plugin = loaded; this.logger.info( { pluginId, methods: ready.methods, providers: ready.providers }, "Loaded plugin", @@ -662,6 +690,65 @@ export class PluginRuntime { return loaded; } + /** + * Frames still in flight belong to the closed session; only a fresh hello + * means the child redialled and wants a new one. Attaching on the close + * instead would leave a socket nobody speaks to until the host's hello + * timeout closed it, and that close would attach another. + */ + private routePluginFrame(input: PluginFrameInput): void { + const { session, frame, isBinary } = input; + if (session.socket.readyState === 1) { + session.socket.receive(frame, isBinary); + return; + } + if (session.reattaching || !isSessionHelloFrame({ frame, isBinary })) return; + if (!this.canReattachPluginSession(input)) return; + session.reattaching = true; + void this.reattachPluginSession(input).finally(() => { + session.reattaching = false; + }); + } + + private async reattachPluginSession(input: PluginFrameInput): Promise { + const { session, pluginId, sessionHost, child, frame, isBinary } = input; + const replacement = new PluginSessionSocket(child); + try { + const attachment = await sessionHost.attachPluginSocket(pluginId, replacement); + if (!this.canReattachPluginSession(input)) { + replacement.close(); + return; + } + session.socket = replacement; + if (session.plugin) { + session.plugin.sessionSocket = replacement; + session.plugin.sessionClosed = attachment.closed; + } + replacement.receive(frame, isBinary); + this.appendLog(pluginId, "stdout", "[paseo] Re-attached plugin session"); + } catch (error) { + const reason = error instanceof Error ? error.message : String(error); + this.appendLog(pluginId, "stderr", `[paseo] Failed to re-attach plugin session: ${reason}`); + this.logger.warn({ pluginId, err: error }, "Failed to re-attach a plugin session"); + this.notify(pluginId, `Plugin session could not be re-attached: ${pluginId}`); + } + } + + /** + * A published plugin whose process is still up. Anything else is either still + * loading or being torn down, and must not get a new session. + */ + private canReattachPluginSession(input: { + session: PluginSessionBinding; + pluginId: string; + child: PluginChild; + }): boolean { + const { session, pluginId, child } = input; + return ( + session.plugin !== null && this.plugins.get(pluginId) === session.plugin && child.connected + ); + } + private handleChildMessage(loaded: LoadedPlugin, message: PluginProcessMessage): void { if (message.type === "hooks.changed") { loaded.hooks = message.hooks; @@ -1068,6 +1155,16 @@ export class PluginRuntime { } } +function isSessionHelloFrame(input: { frame: string | Uint8Array; isBinary: boolean }): boolean { + const { frame, isBinary } = input; + if (isBinary || typeof frame !== "string") return false; + try { + return (JSON.parse(frame) as { type?: unknown }).type === "hello"; + } catch { + return false; + } +} + function readConnectionId(value: unknown): string | null { if (typeof value !== "object" || value === null || Array.isArray(value)) return null; const connectionId = Reflect.get(value, "connectionId"); diff --git a/packages/server/src/server/websocket-server.relay-reconnect.test.ts b/packages/server/src/server/websocket-server.relay-reconnect.test.ts index 79577d6889..15e68e9c03 100644 --- a/packages/server/src/server/websocket-server.relay-reconnect.test.ts +++ b/packages/server/src/server/websocket-server.relay-reconnect.test.ts @@ -18,6 +18,7 @@ import { TerminalStreamOpcode, } from "@getpaseo/protocol/terminal-stream-protocol"; import { CLIENT_CAPS } from "@getpaseo/protocol/client-capabilities"; +import { APPLICATION_SOCKET_LEASE_MS } from "./websocket/physical-socket.js"; type SocketListener = (...args: unknown[]) => void; @@ -546,6 +547,32 @@ describe("relay external socket reconnect behavior", () => { await server.close(); }); + test("keeps a plugin socket whose heartbeat stalls past the application lease", async () => { + const server = createServer(); + const socket = new MockSocket(); + await server.attachPluginSocket("stalled", socket); + socket.emit("message", JSON.stringify(createHelloMessage("plugin:stalled"))); + socket.emit("message", JSON.stringify({ type: "ping" })); + + // Event loop starved past the lease: no further ping arrives. + await vi.advanceTimersByTimeAsync(APPLICATION_SOCKET_LEASE_MS * 2); + + expect(socket.readyState).toBe(1); + await server.close(); + }); + + test("still reaps an ordinary socket whose heartbeat stalls past the application lease", async () => { + const server = createServer(); + const socket = new MockSocket(); + await attachRelayAndHello({ server, socket, clientId: "cid-stalled" }); + socket.emit("message", JSON.stringify({ type: "ping" })); + + await vi.advanceTimersByTimeAsync(APPLICATION_SOCKET_LEASE_MS * 2); + + expect(socket.readyState).toBe(3); + await server.close(); + }); + test("rejects ordinary sockets that claim the reserved plugin client id", async () => { const server = createServer(); const socket = new MockSocket(); diff --git a/packages/server/src/server/websocket-server.ts b/packages/server/src/server/websocket-server.ts index 705a53342e..17d5a03501 100644 --- a/packages/server/src/server/websocket-server.ts +++ b/packages/server/src/server/websocket-server.ts @@ -2203,7 +2203,11 @@ export class VoiceAssistantWebSocketServer { this.recordInboundMessageType(message.type); if (message.type === "ping") { - this.applicationSocketLease.claim(ws); + // A plugin socket is IPC to a child this daemon already supervises, not + // an abandonable application socket. + if (!this.pluginSocketIds.has(ws)) { + this.applicationSocketLease.claim(ws); + } this.sendToClient(ws, { type: "pong" }); return; }