fix(server): stop an expired application lease from permanently killing a plugin session (#4912)

* fix(server): keep plugin sockets out of the application socket lease

A plugin's IPC socket is claimed by the application socket lease as soon as
its client sends its first ping. If that client's heartbeat is then starved
for longer than APPLICATION_SOCKET_LEASE_MS — a blocked event loop, a
SIGSTOP, a sleeping machine — the daemon reaps the socket with "Closing
physical WebSocket with expired application lease".

The plugin never comes back. plugin-process.ts builds its DaemonClient with
`reconnect: { enabled: false }`, so unlike an app or menubar client it cannot
redial, and there is no "Client disconnected; waiting for reconnect" because
no reconnect is ever scheduled. Every later `paseo.*` host API call then
fails with "Transport not connected (status: disconnected)" while
`paseo plugin ls` still reports the plugin as running. Only
`paseo plugin reload <id>` recovers it.

The lease is a backstop for an abandoned application socket. A plugin socket
is IPC to a child this daemon spawned and already supervises through the
child's own close, so heartbeat silence there is not evidence of abandonment
— it only means the child is busy or paused. Reaping it strands a plugin that
was about to resume, which is strictly worse than leaving it attached.

Ordinary sockets are unaffected and still reaped on lease expiry; the second
test pins that down so the exemption cannot widen unnoticed.

Fixes #4883

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* fix(server): let a plugin session recover when the daemon closes its socket

The lease exemption stops the common cause, but any daemon-side close still
stranded a plugin for good: the child's DaemonClient was built with
`reconnect: { enabled: false }`, and even with it enabled the child had
nothing to redial onto, because the runtime attached exactly one
PluginSessionSocket per load and routed frames to that captured const.

Enable reconnect on the plugin client, and have the runtime stand up a
replacement session when a socket closes underneath a live child. Frames now
follow the current socket, and `loaded.sessionSocket` / `sessionClosed` track
the replacement so stop and teardown still await the right attachment. The
session host already supported this — attaching a second socket for the same
plugin id gives it a second, independently owned session.

A plugin on its way out must not be resurrected. Both stop paths
(`stopPluginById`, `stopAll`) clear the catalog entry before they close
anything, so the catalog check is what distinguishes a shutdown from a socket
worth replacing; a dead child is caught by `child.connected`.

The client waits DEFAULT_RECONNECT_BASE_DELAY_MS (1500ms) before redialling
and gives up after DEFAULT_CONNECT_TIMEOUT_MS, so the replacement is attached
long before the first attempt arrives, and a lost race would retry anyway.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* refactor(server): trim the comments on the plugin session fix

The reasoning lives in the commit messages and the PR. Leave only what the
code cannot say for itself: why a plugin socket is not an application socket,
and why the catalog check is what distinguishes a shutdown.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* fix(server): re-attach a plugin session on the child's redial, not on close

Attaching eagerly when the socket closed put a socket in place that nobody
was speaking to yet. Two ways that goes wrong, both reachable under exactly
the conditions #4883 describes - a child whose event loop is starved:

- Nothing sends a hello on the replacement, so the host closes it after
  HELLO_TIMEOUT_MS (websocket-server.ts:1310), and that close attached
  another. A wedged child drives that cycle until it exits, filling the
  plugin's 500-entry log ring with re-attach lines.
- Frames the child had already queued arrive after the close and land on the
  replacement while it is still pending, which closes it with
  WS_CLOSE_INVALID_HELLO (websocket-server.ts:2151) and attaches another.

Route frames through the current socket instead, and stand a replacement up
only when a hello arrives while the old one is closed - the one point that
can tell leftovers of the dead session from a genuine redial, since the IPC
channel carries no session id. A starved child now attaches nothing until it
wakes up and redials.

Also re-check the guards after the attach await and close the replacement if
they no longer hold, and surface a failed re-attach through appendLog and
notify rather than only a daemon log line, so a plugin left without a session
is visible instead of silently repeating the original bug.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* refactor(server): follow the server style guide and drop test timing

Two review findings against packages/server/CLAUDE.md.

Helper shape (CLAUDE.md: `function` declarations over arrow assignments,
object parameters when a function takes more than one argument). The
re-attach helpers were arrow assignments closing over `loadPlugin` locals, and
two took `(frame, isBinary)` positionally. They are now private methods taking
a single object. The mutable session state they shared - the current socket,
the in-flight flag, the loaded plugin - moves into a PluginSessionBinding the
methods read, which is what made the closures hard to lift out before.

Test determinism (CLAUDE.md: no timing/randomness). The reconnect tests waited
on fixed 500ms sleeps and 15s vi.waitFor polling windows, so they inferred
lifecycle state from elapsed time. The tracked session host now exposes
waitForHellos, resolving on the handshake itself, and the lazy-attach case
runs against a stub child that is driven frame by frame: close, then a stale
frame that must not attach, then a hello that must. No sleeps and no polling
remain in these tests, and the file's suite drops from ~40s to ~11s.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* test(plugins): verify host API and observation recovery

* test(plugins): await connection and catalog events

---------

Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
Co-authored-by: Mohamed Boudra <boudra.moha@gmail.com>

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",

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<void> {
     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<number, (id: string) => void>();
+  const nameWaiters = new Map<string, (name: string) => void>();
+  let reconnected = Promise.resolve();
+  let release: (() => Promise<void>) | 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<string>((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<string>((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<void>((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);

packages/server/src/server/plugins/plugin-process.ts

@@ -260,7 +260,8 @@ async function initialize(message: Extract<PluginProcessRequest, { type: "initia
     clientId: createPluginClientId(message.pluginId),
     clientType: "cli",
     appVersion: message.appVersion,
-    reconnect: { enabled: false },
+    // The runtime re-attaches a session when the daemon drops this socket.
+    reconnect: { enabled: true },
     transportFactory,
   });
   paseo = createPaseoApi(daemonClient);

packages/server/src/server/plugins/runtime.posix.test.ts

@@ -124,8 +124,28 @@ function createTestRuntime(
 
 function createTrackedSessionHost() {
   const active = new Set<object>();
+  // 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<void> {
+      if (hellos.length >= count) return Promise.resolve();
+      return new Promise<void>((resolve) => {
+        waiters.add({ count, resolve });
+      });
+    },
     host: {
       async attachPluginSocket(_pluginId: string, socket: PluginSessionSocket) {
         const closed = new Promise<void>((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<void>((resolve) => {
+      finishAttachment = resolve;
+    });
+    let replacementAttached!: (socket: PluginSessionSocket) => void;
+    const replacementStarted = new Promise<PluginSessionSocket>((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<void>((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();
+    }
+  });
 });

packages/server/src/server/plugins/runtime.ts

@@ -74,6 +74,22 @@ interface LoadedPlugin {
   sessionClosed: Promise<void> | 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<string, PendingInvocation>();
     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<void> {
+    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");

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();

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;
       }