Make local plugins manageable and debuggable (#3446)

* feat(plugins): expose Paseo API to local plugins

Plugin surfaces borrow the selected host connection while subprocess handlers receive an exclusively owned IPC session, so plugins can use normal workspace and agent operations without inventing another transport.

* feat(plugins): reload global enablement safely

Treat the global plugin switch as a live daemon setting and teach plugin automation to inspect it before acting. Enabling remains permission-gated because plugin backend code can access the daemon machine and client contributions execute inside the app.

* feat(plugins): expose plugin logs

Capture plugin stdout and stderr separately from IPC so authors can debug initialization, RPC, cleanup, and crash behavior from the app or CLI. Keep a bounded tail across lifecycle recovery while preserving structured output in the daemon log.

* fix(build): order plugin dependency builds

* test(plugins): align startup and CLI expectations

packages/server/src/server/bootstrap.smoke.test.ts

@@ -48,16 +48,6 @@ type WebSocketProbeResult =
   | { status: "connected" }
   | { status: "rejected"; statusCode: number | null };
 
-function processExists(pid: number): boolean {
-  try {
-    process.kill(pid, 0);
-    return true;
-  } catch (error) {
-    if ((error as NodeJS.ErrnoException).code === "ESRCH") return false;
-    throw error;
-  }
-}
-
 describe("paseo daemon bootstrap", () => {
   afterEach(() => {
     vi.restoreAllMocks();
@@ -593,7 +583,7 @@ describe("paseo daemon bootstrap", () => {
     }
   });
 
-  test("rolls back already-open standalone listener when main daemon listen fails", async () => {
+  test("rolls back standalone listener without starting plugins when main listen fails", async () => {
     const mainPort = await findFreePort();
     const standalonePort = await findFreePort();
     const occupiedMain = http.createServer((_req, res) => {
@@ -648,10 +638,7 @@ export default function contribute(plugin: unknown) {
       await expect(daemon.start()).rejects.toThrow();
       await expect(fetch(`http://127.0.0.1:${standalonePort}/api/health`)).rejects.toThrow();
       if (!isPlatform("win32")) {
-        const pluginPid = Number(await readFile(pluginPidPath, "utf8"));
-        await vi.waitFor(() => {
-          expect(processExists(pluginPid)).toBe(false);
-        });
+        await expect(readFile(pluginPidPath, "utf8")).rejects.toMatchObject({ code: "ENOENT" });
       }
     } finally {
       await daemon.stop().catch(() => undefined);

packages/server/src/server/bootstrap.ts

@@ -1491,7 +1491,6 @@ export async function createPaseoDaemon(
   const start = async () => {
     let mainStarted = false;
     try {
-      await pluginRuntime.start();
       if (serviceProxyListenTarget) {
         const boundServiceProxyTarget = await serviceProxy.startStandalone({
           listenTarget: serviceProxyListenTarget,
@@ -1580,6 +1579,7 @@ export async function createPaseoDaemon(
                 getHostnames: () => configuredHostnames,
                 daemonStatusRpc: dependencies.serverFeatureOverrides?.daemonStatusRpc,
                 relayConfig: dependencies.serverFeatureOverrides?.relayConfig,
+                startPaused: true,
               },
               workspaceAutoName,
               config.auth,
@@ -1632,6 +1632,9 @@ export async function createPaseoDaemon(
               workspaceSetupRuntime,
               pluginRuntime,
             );
+            pluginRuntime.bindPaseoSessionHost(wsServer);
+            await pluginRuntime.start();
+            wsServer.beginAcceptingConnections();
             relayRuntime = createRelayRuntime({
               config: {
                 enabled: relayEnabled,

packages/server/src/server/daemon-config-store.test.ts

@@ -36,6 +36,8 @@ function reloadableConfig(
       maxProcessConcurrency: git.maxProcessConcurrency ?? 8,
     },
     app: { baseUrl: "https://app.paseo.sh" },
+    pluginsEnabled: persisted.pluginsEnabled ?? false,
+    plugins: persisted.plugins ?? {},
   };
 }
 
@@ -940,6 +942,31 @@ describe("DaemonConfigStore reload", () => {
     expect(store.get().git).toEqual({ maxProcessesPerSecond: 12, maxProcessConcurrency: 3 });
   });
 
+  test("applies the global plugin switch in both directions", () => {
+    const { paseoHome, store, persisted } = createReloadableStore({
+      initialPersisted: { version: 1, pluginsEnabled: false },
+    });
+    const changes: unknown[] = [];
+    store.onFieldChange("pluginsEnabled", (value) => changes.push(value));
+
+    writeConfig(paseoHome, { ...persisted, pluginsEnabled: true });
+    expect(store.reload()).toEqual({
+      appliedPaths: ["pluginsEnabled"],
+      restartRequiredPaths: [],
+      overrideControlledPaths: [],
+    });
+    expect(store.get().pluginsEnabled).toBe(true);
+
+    writeConfig(paseoHome, { ...persisted, pluginsEnabled: false });
+    expect(store.reload()).toEqual({
+      appliedPaths: ["pluginsEnabled"],
+      restartRequiredPaths: [],
+      overrideControlledPaths: [],
+    });
+    expect(store.get().pluginsEnabled).toBe(false);
+    expect(changes).toEqual([true, false]);
+  });
+
   test("classifies every leaf when a parent subtree is added", () => {
     const { paseoHome, store } = createReloadableStore({
       initialPersisted: { version: 1 },

packages/server/src/server/plugins/daemon-transport.test.ts

@@ -0,0 +1,36 @@
+import { describe, expect, it } from "vitest";
+import { createPluginDaemonTransportFactory } from "./daemon-transport.js";
+import type { PluginProcessMessage, PluginProcessRequest } from "./plugin-process-protocol.js";
+
+describe("plugin daemon IPC transport", () => {
+  it("preserves incoming text and binary frame order", async () => {
+    let receive: ((message: PluginProcessRequest) => void) | null = null;
+    const sent: PluginProcessMessage[] = [];
+    const factory = createPluginDaemonTransportFactory({
+      send(message) {
+        sent.push(message);
+      },
+      onMessage(handler) {
+        receive = handler;
+        return () => {
+          receive = null;
+        };
+      },
+    });
+    const transport = factory({ url: "ipc://plugin" });
+    const frames: Array<{ data: unknown; isBinary: boolean }> = [];
+    transport.onMessage((data, isBinary) => frames.push({ data, isBinary }));
+    const opened = new Promise<void>((resolve) => transport.onOpen(resolve));
+
+    await opened;
+    receive?.({ type: "paseo_frame", data: "first", isBinary: false });
+    receive?.({ type: "paseo_frame", data: new Uint8Array([7, 8]), isBinary: true });
+    transport.send("outbound");
+
+    expect(frames).toEqual([
+      { data: "first", isBinary: false },
+      { data: new Uint8Array([7, 8]), isBinary: true },
+    ]);
+    expect(sent).toEqual([{ type: "paseo_frame", data: "outbound", isBinary: false }]);
+  });
+});

packages/server/src/server/plugins/daemon-transport.ts

@@ -0,0 +1,72 @@
+import type { DaemonTransportFactory } from "@getpaseo/client/internal/daemon-client-transport-types";
+import type { PluginProcessMessage, PluginProcessRequest } from "./plugin-process-protocol.js";
+
+interface PluginProcessPort {
+  send(message: PluginProcessMessage): void;
+  onMessage(handler: (message: PluginProcessRequest) => void): () => void;
+}
+
+function normalizeFrame(data: string | Uint8Array | ArrayBuffer): string | Uint8Array {
+  if (typeof data === "string") return data;
+  if (data instanceof Uint8Array) {
+    return new Uint8Array(data.buffer, data.byteOffset, data.byteLength);
+  }
+  return new Uint8Array(data);
+}
+
+export function createPluginDaemonTransportFactory(
+  port: PluginProcessPort,
+): DaemonTransportFactory {
+  return () => {
+    const messageHandlers = new Set<(data: unknown, isBinary: boolean) => void>();
+    const openHandlers = new Set<() => void>();
+    const closeHandlers = new Set<(event?: unknown) => void>();
+    let isOpen = true;
+    const unsubscribe = port.onMessage((message) => {
+      if (message.type === "paseo_frame") {
+        for (const handler of messageHandlers) handler(message.data, message.isBinary);
+      }
+      if (message.type === "paseo_close") closeTransport();
+    });
+
+    function closeTransport(): void {
+      if (!isOpen) return;
+      isOpen = false;
+      unsubscribe();
+      for (const handler of closeHandlers) handler();
+    }
+
+    queueMicrotask(() => {
+      if (!isOpen) return;
+      for (const handler of openHandlers) handler();
+    });
+
+    return {
+      send(data) {
+        if (!isOpen) throw new Error("Plugin daemon transport is closed");
+        const frame = normalizeFrame(data);
+        port.send({ type: "paseo_frame", data: frame, isBinary: typeof frame !== "string" });
+      },
+      close() {
+        if (!isOpen) return;
+        port.send({ type: "paseo_close" });
+        closeTransport();
+      },
+      onMessage(handler) {
+        messageHandlers.add(handler);
+        return () => messageHandlers.delete(handler);
+      },
+      onOpen(handler) {
+        openHandlers.add(handler);
+        return () => openHandlers.delete(handler);
+      },
+      onClose(handler) {
+        closeHandlers.add(handler);
+        return () => closeHandlers.delete(handler);
+      },
+      onError() {
+        return () => undefined;
+      },
+    };
+  };
+}

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

@@ -38,7 +38,46 @@ function createService(
   plugins: Record<string, { source: "directory"; path: string; enabled?: boolean }> = {},
   runtime?: ConstructorParameters<typeof PluginService>[2],
 ): PluginService {
-  return new PluginService(pino({ level: "silent" }), createStore(home, plugins), runtime);
+  return bindTestSessionHost(
+    new PluginService(pino({ level: "silent" }), createStore(home, plugins), runtime),
+  );
+}
+
+function bindTestSessionHost(service: PluginService): PluginService {
+  service.bindPaseoSessionHost({
+    async attachPluginSocket(_pluginId, socket) {
+      const closed = new Promise<void>((resolve) => socket.once("close", resolve));
+      socket.on("message", (data) => {
+        if (typeof data !== "string") return;
+        const message = JSON.parse(data);
+        if (message.type !== "hello") return;
+        socket.send(
+          JSON.stringify({
+            type: "session",
+            message: {
+              type: "status",
+              payload: {
+                status: "server_info",
+                serverId: "plugin-service-test",
+                hostname: "plugin-service-test",
+                version: "0.4.0",
+                features: {},
+              },
+            },
+          }),
+        );
+      });
+      return { closed };
+    },
+  });
+  return service;
+}
+
+function catalogIds(service: PluginService): string[] {
+  return service
+    .catalog()
+    .map(({ id }) => id)
+    .sort();
 }
 
 afterEach(async () => {
@@ -58,6 +97,8 @@ function createPausedRuntime() {
   const runtime: NonNullable<ConstructorParameters<typeof PluginService>[2]> = {
     catalog: () => [...running].map((id) => ({ id, clientBundle: "bundle" })),
     invoke: async () => undefined,
+    getLogs: () => [],
+    clearLogs: () => undefined,
     startPlugin: async (pluginId, _path, canPublish) => {
       markStarted();
       await startGate;
@@ -69,6 +110,7 @@ function createPausedRuntime() {
       running.clear();
     },
     subscribe: () => () => undefined,
+    bindPaseoSessionHost: () => undefined,
   };
   return { runtime, started, releaseStart };
 }
@@ -87,6 +129,8 @@ function createPluginSelectivePausedRuntime(pausedPluginId: string) {
   const runtime: NonNullable<ConstructorParameters<typeof PluginService>[2]> = {
     catalog: () => [...running].map((id) => ({ id, clientBundle: "bundle" })),
     invoke: async () => undefined,
+    getLogs: () => [],
+    clearLogs: () => undefined,
     startPlugin: async (pluginId, _path, canPublish) => {
       starts.push(pluginId);
       if (pluginId === pausedPluginId) {
@@ -101,11 +145,78 @@ function createPluginSelectivePausedRuntime(pausedPluginId: string) {
       running.clear();
     },
     subscribe: () => () => undefined,
+    bindPaseoSessionHost: () => undefined,
   };
   return { runtime, started, releaseStart, starts };
 }
 
 describe("PluginService", () => {
+  it("retains logs when disabled and clears them only when removed", async () => {
+    const home = await mkdtemp(path.join(tmpdir(), "paseo-plugin-home-"));
+    roots.push(home);
+    const entries = [
+      {
+        sequence: 1,
+        timestamp: "2026-08-16T12:00:00.000Z",
+        stream: "stdout" as const,
+        message: "ready",
+      },
+    ];
+    const cleared: string[] = [];
+    const runtime: NonNullable<ConstructorParameters<typeof PluginService>[2]> = {
+      catalog: () => [],
+      invoke: async () => undefined,
+      getLogs: () => entries,
+      clearLogs: (pluginId) => {
+        cleared.push(pluginId);
+        entries.length = 0;
+      },
+      startPlugin: async () => undefined,
+      stopPluginById: async () => false,
+      stopAll: async () => undefined,
+      subscribe: () => () => undefined,
+      bindPaseoSessionHost: () => undefined,
+    };
+    const service = createService(
+      home,
+      { example: { source: "directory", path: "/plugins/example", enabled: false } },
+      runtime,
+    );
+
+    expect(service.getLogs("example")).toEqual(entries);
+    await service.disablePlugin("example");
+    expect(service.getLogs("example")).toEqual(entries);
+    expect(cleared).toEqual([]);
+
+    await service.removePlugin("example");
+    expect(cleared).toEqual(["example"]);
+    expect(entries).toEqual([]);
+  });
+
+  it("publishes each configured plugin after its startup state settles", async () => {
+    const home = await mkdtemp(path.join(tmpdir(), "paseo-plugin-home-"));
+    roots.push(home);
+    const first = await createPlugin(
+      "startup-first",
+      `export default function contribute(plugin: unknown) { void plugin; return () => undefined; }`,
+    );
+    const second = await createPlugin(
+      "startup-second",
+      `export default function contribute(plugin: unknown) { void plugin; return () => undefined; }`,
+    );
+    const service = createService(home, {
+      "startup-first": { source: "directory", path: first },
+      "startup-second": { source: "directory", path: second },
+    });
+    const snapshots: string[][] = [];
+    service.subscribe(() => snapshots.push(catalogIds(service)));
+
+    await service.start();
+
+    expect(snapshots).toEqual([["startup-first"], ["startup-first", "startup-second"]]);
+    await service.stopAllPlugins();
+  }, 20_000);
+
   it("uses an explicit config key, exposes reload failure, and retries from disk", async () => {
     const home = await mkdtemp(path.join(tmpdir(), "paseo-plugin-home-"));
     roots.push(home);
@@ -114,7 +225,7 @@ describe("PluginService", () => {
       `export default function contribute(plugin: unknown) { void plugin; return () => undefined; }`,
     );
     const store = createStore(home);
-    const service = new PluginService(pino({ level: "silent" }), store);
+    const service = bindTestSessionHost(new PluginService(pino({ level: "silent" }), store));
     await service.start();
 
     await expect(
@@ -140,7 +251,7 @@ describe("PluginService", () => {
     );
     await expect(service.reloadPlugin("work-plugin")).resolves.toMatchObject({ status: "running" });
     await service.stopAllPlugins();
-  });
+  }, 20_000);
 
   it("disables and removes a plugin without touching its source directory", async () => {
     const home = await mkdtemp(path.join(tmpdir(), "paseo-plugin-home-"));
@@ -155,7 +266,7 @@ export default function contribute(plugin: unknown) {
 }`,
     );
     const store = createStore(home);
-    const service = new PluginService(pino({ level: "silent" }), store);
+    const service = bindTestSessionHost(new PluginService(pino({ level: "silent" }), store));
     await service.start();
     await service.installDirectory({ path: directory });
 
@@ -168,7 +279,7 @@ export default function contribute(plugin: unknown) {
     expect(service.listPlugins()).toEqual([]);
     await expect(stat(directory)).resolves.toMatchObject({});
     await service.stopAllPlugins();
-  });
+  }, 20_000);
 
   it("detaches every plugin synchronously when the global switch turns off and recovers", async () => {
     const home = await mkdtemp(path.join(tmpdir(), "paseo-plugin-home-"));
@@ -182,7 +293,7 @@ export default function contribute(plugin: unknown) {
       `export default function contribute(plugin: unknown) { void plugin; return () => undefined; }`,
     );
     const store = createStore(home);
-    const service = new PluginService(pino({ level: "silent" }), store);
+    const service = bindTestSessionHost(new PluginService(pino({ level: "silent" }), store));
     await service.start();
     await service.installDirectory({ path: first });
     await service.installDirectory({ path: second });
@@ -200,7 +311,7 @@ export default function contribute(plugin: unknown) {
       { id: "second", status: "running" },
     ]);
     await service.stopAllPlugins();
-  });
+  }, 20_000);
 
   it("does not publish an in-flight start after a later global disable", async () => {
     const home = await mkdtemp(path.join(tmpdir(), "paseo-plugin-home-"));

packages/server/src/server/plugins/index.ts

@@ -2,6 +2,7 @@ import path from "node:path";
 import type pino from "pino";
 import {
   PluginIdSchema,
+  type PluginLogEntry,
   type PluginListItem,
   type PluginSource,
 } from "@getpaseo/protocol/messages";
@@ -12,10 +13,13 @@ import { PluginRuntime } from "./runtime.js";
 interface PluginRuntimePort {
   catalog(): Array<{ id: string; clientBundle: string }>;
   invoke(pluginId: string, method: string, input: unknown): Promise<unknown>;
+  getLogs(pluginId: string): PluginLogEntry[];
+  clearLogs(pluginId: string): void;
   startPlugin(pluginId: string, path: string, canPublish: () => boolean): Promise<void>;
   stopPluginById(pluginId: string): Promise<boolean>;
   stopAll(): Promise<void>;
   subscribe(listener: (pluginId: string, error?: string) => void): () => void;
+  bindPaseoSessionHost(sessionHost: Parameters<PluginRuntime["bindPaseoSessionHost"]>[0]): void;
 }
 
 export class PluginService {
@@ -45,6 +49,10 @@ export class PluginService {
     return () => this.listeners.delete(listener);
   }
 
+  bindPaseoSessionHost(sessionHost: Parameters<PluginRuntime["bindPaseoSessionHost"]>[0]): void {
+    this.runtime.bindPaseoSessionHost(sessionHost);
+  }
+
   async start(): Promise<void> {
     if (this.started) return;
     this.started = true;
@@ -54,6 +62,7 @@ export class PluginService {
       for (const [pluginId, source] of Object.entries(config.plugins ?? {})) {
         if (source.enabled === false) continue;
         await this.startConfigured(pluginId);
+        this.notify(pluginId);
       }
     }
     this.configStore.onFieldChange("pluginsEnabled", (value) => {
@@ -83,6 +92,11 @@ export class PluginService {
       .sort((left, right) => left.id.localeCompare(right.id));
   }
 
+  getLogs(pluginId: string): PluginLogEntry[] {
+    this.requireSource(pluginId);
+    return this.runtime.getLogs(pluginId);
+  }
+
   catalog(): Array<{ id: string; clientBundle: string }> {
     return this.runtime.catalog();
   }
@@ -164,6 +178,7 @@ export class PluginService {
     this.configStore.patch({ plugins: sources });
     await this.enqueue(async () => {
       await stopping;
+      this.runtime.clearLogs(pluginId);
       this.errors.delete(pluginId);
       this.notify(pluginId);
     });

packages/server/src/server/plugins/plugin-paseo-api.e2e.test.ts

@@ -0,0 +1,165 @@
+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 { DaemonClient } from "../test-utils/daemon-client.js";
+import { createTestPaseoDaemon } from "../test-utils/paseo-daemon.js";
+
+const roots: string[] = [];
+
+afterEach(async () => {
+  await Promise.all(roots.splice(0).map((root) => rm(root, { recursive: true, force: true })));
+});
+
+test("plugin handlers create workspaces and agents through their Paseo API", async () => {
+  const pluginDirectory = await mkdtemp(path.join(tmpdir(), "paseo-api-plugin-"));
+  const workspaceDirectory = await mkdtemp(path.join(tmpdir(), "paseo-api-workspace-"));
+  roots.push(pluginDirectory, workspaceDirectory);
+  await writeFile(
+    path.join(pluginDirectory, "paseo-plugin.json"),
+    JSON.stringify({ id: "paseo-api" }),
+  );
+  await writeFile(
+    path.join(pluginDirectory, "index.tsx"),
+    `import { defineRpc, type PluginContext } from "@paseo/plugin";
+import { z } from "zod";
+
+const create = defineRpc({
+  name: "create",
+  input: z.object({ path: z.string() }),
+  output: z.object({ workspaceId: z.string(), agentId: z.string() }),
+});
+
+export default function contribute(plugin: PluginContext) {
+  plugin.handle(create, async ({ path }, { paseo }) => {
+    const workspace = await paseo.workspaces.create({
+      source: { kind: "directory", path },
+      title: "Plugin workspace",
+    });
+    const agent = await workspace.agents.create({
+      config: { provider: "codex/test" },
+      prompt: "Created by a plugin handler",
+    });
+    return { workspaceId: workspace.id, agentId: agent.id };
+  });
+  return () => undefined;
+}`,
+  );
+
+  const daemon = await createTestPaseoDaemon();
+  const client = new DaemonClient({
+    url: `ws://127.0.0.1:${daemon.port}/ws`,
+    appVersion: "0.4.0",
+  });
+
+  try {
+    await client.connect();
+    await client.patchDaemonConfig({ pluginsEnabled: true });
+    await expect(client.installDirectoryPlugin(pluginDirectory)).resolves.toMatchObject({
+      id: "paseo-api",
+      status: "running",
+    });
+
+    const created = await client.invokePluginRpc("paseo-api", "create", {
+      path: workspaceDirectory,
+    });
+
+    expect(created).toEqual({
+      workspaceId: expect.stringMatching(/^wks_/),
+      agentId: expect.stringMatching(/^[0-9a-f-]{36}$/),
+    });
+    if (typeof created !== "object" || created === null) {
+      throw new Error("Plugin returned an invalid creation result");
+    }
+    await client.removePlugin("paseo-api");
+    const workspaces = await client.fetchWorkspaces();
+    const agents = await client.fetchAgents();
+    expect(workspaces.entries.map((workspace) => workspace.id)).toContain(
+      Reflect.get(created, "workspaceId"),
+    );
+    expect(agents.entries.map((entry) => entry.agent.id)).toContain(
+      Reflect.get(created, "agentId"),
+    );
+  } finally {
+    await client.close().catch(() => undefined);
+    await daemon.close();
+  }
+}, 60_000);
+
+test("daemon config reload enables and disables configured plugins without restarting", async () => {
+  const pluginDirectory = await mkdtemp(path.join(tmpdir(), "paseo-reload-plugin-"));
+  const paseoHomeRoot = await mkdtemp(path.join(tmpdir(), "paseo-reload-home-"));
+  const paseoHome = path.join(paseoHomeRoot, ".paseo");
+  roots.push(pluginDirectory, paseoHomeRoot);
+  await writeFile(
+    path.join(pluginDirectory, "paseo-plugin.json"),
+    JSON.stringify({ id: "reloadable-plugin" }),
+  );
+  await writeFile(
+    path.join(pluginDirectory, "index.tsx"),
+    `export default function contribute(plugin: unknown) {
+  void plugin;
+  return () => undefined;
+    }`,
+  );
+
+  const plugins = {
+    "reloadable-plugin": { source: "directory" as const, path: pluginDirectory, enabled: true },
+  };
+  await mkdir(paseoHome, { recursive: true });
+  await writeFile(
+    path.join(paseoHome, "config.json"),
+    `${JSON.stringify({ version: 1, pluginsEnabled: false, plugins }, null, 2)}\n`,
+  );
+  const daemon = await createTestPaseoDaemon({
+    paseoHomeRoot,
+    cleanup: false,
+    pluginsEnabled: false,
+    plugins,
+  });
+  const client = new DaemonClient({
+    url: `ws://127.0.0.1:${daemon.port}/ws`,
+    appVersion: "0.4.0",
+  });
+  const configPath = path.join(daemon.paseoHome, "config.json");
+
+  async function setPluginsEnabled(enabled: boolean): Promise<void> {
+    const config = JSON.parse(await readFile(configPath, "utf8"));
+    await writeFile(
+      configPath,
+      `${JSON.stringify({ ...config, pluginsEnabled: enabled }, null, 2)}\n`,
+    );
+  }
+
+  try {
+    await client.connect();
+    await expect(client.listPlugins()).resolves.toEqual([
+      expect.objectContaining({ id: "reloadable-plugin", status: "disabled" }),
+    ]);
+
+    await setPluginsEnabled(true);
+    await expect(client.reloadDaemonConfig()).resolves.toMatchObject({
+      requestId: expect.any(String),
+      appliedPaths: expect.arrayContaining(["pluginsEnabled"]),
+      restartRequiredPaths: [],
+      overrideControlledPaths: [],
+    });
+    await expect
+      .poll(async () => (await client.listPlugins()).find(({ id }) => id === "reloadable-plugin"))
+      .toMatchObject({ enabled: true, status: "running" });
+
+    await setPluginsEnabled(false);
+    await expect(client.reloadDaemonConfig()).resolves.toEqual({
+      requestId: expect.any(String),
+      appliedPaths: ["pluginsEnabled"],
+      restartRequiredPaths: [],
+      overrideControlledPaths: [],
+    });
+    await expect
+      .poll(async () => (await client.listPlugins()).find(({ id }) => id === "reloadable-plugin"))
+      .toMatchObject({ enabled: true, status: "disabled" });
+  } finally {
+    await client.close().catch(() => undefined);
+    await daemon.close();
+  }
+}, 60_000);

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

@@ -1,10 +1,14 @@
 export type PluginProcessRequest =
-  | { type: "initialize"; bundle: string }
+  | { type: "initialize"; pluginId: string; bundle: string }
   | { type: "invoke"; requestId: string; method: string; input: unknown }
-  | { type: "shutdown" };
+  | { type: "shutdown" }
+  | { type: "paseo_frame"; data: string | Uint8Array; isBinary: boolean }
+  | { type: "paseo_close" };
 
 export type PluginProcessMessage =
   | { type: "ready"; methods: string[] }
   | { type: "result"; requestId: string; output: unknown }
   | { type: "error"; requestId: string; error: string }
-  | { type: "fatal"; error: string };
+  | { type: "fatal"; error: string }
+  | { type: "paseo_frame"; data: string | Uint8Array; isBinary: boolean }
+  | { type: "paseo_close" };

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

@@ -1,8 +1,16 @@
 import type { PluginProcessMessage, PluginProcessRequest } from "./plugin-process-protocol.js";
 import { createRequire } from "node:module";
-import { defineAttachmentSource, defineRpc, type PluginRpcContract } from "@paseo/plugin";
+import {
+  defineAttachmentSource,
+  defineRpc,
+  type PluginHandlerContext,
+  type PluginRpcContract,
+} from "@paseo/plugin";
+import { createPaseoApi, type PaseoApi } from "@getpaseo/client";
+import { DaemonClient } from "@getpaseo/client/internal/daemon-client";
+import { createPluginDaemonTransportFactory } from "./daemon-transport.js";
 
-type RpcHandler = (input: unknown) => unknown | Promise<unknown>;
+type RpcHandler = (input: unknown, context: PluginHandlerContext) => unknown | Promise<unknown>;
 
 interface RegisteredRpc {
   contract: PluginRpcContract;
@@ -10,13 +18,26 @@ interface RegisteredRpc {
 }
 
 const handlers = new Map<string, RegisteredRpc>();
-let cleanup: (() => void) | null = null;
+let cleanup: (() => void | Promise<void>) | null = null;
+let daemonClient: DaemonClient | null = null;
+let paseo: PaseoApi | null = null;
+let stopping = false;
 const nodeRequire = createRequire(import.meta.url);
 
 function send(message: PluginProcessMessage): void {
   process.send?.(message);
 }
 
+function sendAndWait(message: PluginProcessMessage): Promise<void> {
+  return new Promise((resolve) => {
+    if (!process.send) {
+      resolve();
+      return;
+    }
+    process.send(message, () => resolve());
+  });
+}
+
 function describeError(error: unknown): string {
   return error instanceof Error ? error.message : String(error);
 }
@@ -62,29 +83,59 @@ function evaluateBundle(bundle: string): void {
   cleanup = contributedCleanup;
 }
 
+const transportFactory = createPluginDaemonTransportFactory({
+  send,
+  onMessage(handler) {
+    process.on("message", handler);
+    return () => process.off("message", handler);
+  },
+});
+
+async function initialize(message: Extract<PluginProcessRequest, { type: "initialize" }>) {
+  daemonClient = new DaemonClient({
+    url: `ipc://plugin/${encodeURIComponent(message.pluginId)}`,
+    clientId: `plugin:${message.pluginId}`,
+    clientType: "cli",
+    reconnect: { enabled: false },
+    transportFactory,
+  });
+  paseo = createPaseoApi(daemonClient);
+  await daemonClient.connect();
+  evaluateBundle(message.bundle);
+  send({ type: "ready", methods: [...handlers.keys()].sort() });
+}
+
+async function shutdown(): Promise<void> {
+  if (stopping) return;
+  stopping = true;
+  const currentCleanup = cleanup;
+  cleanup = null;
+  try {
+    await currentCleanup?.();
+  } catch (error) {
+    console.error("Plugin cleanup failed", error);
+  }
+  await daemonClient?.close().catch(() => undefined);
+  await sendAndWait({ type: "paseo_close" });
+  daemonClient = null;
+  paseo = null;
+  process.disconnect();
+}
+
 process.on("message", (message: PluginProcessRequest) => {
   if (message.type === "initialize") {
-    try {
-      evaluateBundle(message.bundle);
-      send({ type: "ready", methods: [...handlers.keys()].sort() });
-    } catch (error) {
+    void initialize(message).catch(async (error) => {
       send({ type: "fatal", error: describeError(error) });
-    }
+      await daemonClient?.close().catch(() => undefined);
+    });
     return;
   }
   if (message.type === "shutdown") {
-    const currentCleanup = cleanup;
-    cleanup = null;
-    if (currentCleanup) {
-      try {
-        currentCleanup();
-      } catch (error) {
-        console.error("Plugin cleanup failed", error);
-      }
-    }
-    process.disconnect();
+    void shutdown();
     return;
   }
+  if (message.type === "paseo_frame" || message.type === "paseo_close") return;
+  if (stopping) return;
   const registered = handlers.get(message.method);
   if (!registered) {
     send({
@@ -96,7 +147,10 @@ process.on("message", (message: PluginProcessRequest) => {
   }
   void registered.contract.input
     .parseAsync(message.input)
-    .then(registered.handler)
+    .then((input) => {
+      if (!paseo) throw new Error("Plugin Paseo API is unavailable");
+      return registered.handler(input, { paseo });
+    })
     .then((output) => registered.contract.output.parseAsync(output))
     .then(
       (output) => send({ type: "result", requestId: message.requestId, output }),

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

@@ -1,10 +1,12 @@
-import { mkdtemp, rm, writeFile } from "node:fs/promises";
+import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
 import { tmpdir } from "node:os";
+import { PassThrough } from "node:stream";
 import path from "node:path";
 import { fileURLToPath } from "node:url";
 import pino from "pino";
-import { afterEach, describe, expect, it } from "vitest";
+import { afterEach, describe, expect, it, vi } from "vitest";
 import { PluginRuntime } from "./runtime.js";
+import type { PluginSessionSocket } from "./session-socket.js";
 
 const temporaryDirectories: string[] = [];
 
@@ -22,6 +24,8 @@ function createReloadChild(name: string, events: string[], methods: string[] = [
     for (const listener of listeners.get(event) ?? []) listener(message as never);
   };
   return {
+    stdout: new PassThrough(),
+    stderr: new PassThrough(),
     connected: true,
     killed: false,
     send(message: { type: string }, callback?: (error: Error | null) => void) {
@@ -58,6 +62,84 @@ function createReloadChild(name: string, events: string[], methods: string[] = [
   };
 }
 
+function createTestRuntime(
+  dependencies: NonNullable<ConstructorParameters<typeof PluginRuntime>[1]> = {},
+  logger = pino({ level: "silent" }),
+): PluginRuntime {
+  return new PluginRuntime(logger, {
+    ...dependencies,
+    sessionHost: dependencies.sessionHost ?? {
+      async attachPluginSocket(_pluginId, socket) {
+        const closed = new Promise<void>((resolve) => socket.once("close", resolve));
+        socket.on("message", (data) => {
+          if (typeof data !== "string") return;
+          const message = JSON.parse(data);
+          if (message.type !== "hello") return;
+          socket.send(
+            JSON.stringify({
+              type: "session",
+              message: {
+                type: "status",
+                payload: {
+                  status: "server_info",
+                  serverId: "plugin-test",
+                  hostname: "plugin-test",
+                  version: "0.4.0",
+                  features: {},
+                },
+              },
+            }),
+          );
+        });
+        return { closed };
+      },
+    },
+  });
+}
+
+function createTrackedSessionHost() {
+  const active = new Set<object>();
+  return {
+    active,
+    host: {
+      async attachPluginSocket(_pluginId: string, socket: PluginSessionSocket) {
+        const closed = new Promise<void>((resolve) => socket.once("close", resolve));
+        active.add(socket);
+        socket.once("close", () => active.delete(socket));
+        socket.on("message", (data) => {
+          if (typeof data !== "string") return;
+          const message = JSON.parse(data);
+          if (message.type !== "hello") return;
+          socket.send(
+            JSON.stringify({
+              type: "session",
+              message: {
+                type: "status",
+                payload: {
+                  status: "server_info",
+                  serverId: "tracked-plugin-test",
+                  hostname: "tracked-plugin-test",
+                  version: "0.4.0",
+                  features: {},
+                },
+              },
+            }),
+          );
+        });
+        return { closed };
+      },
+    },
+  };
+}
+
+function lifecycleMessages(runtime: PluginRuntime): string[] {
+  return runtime
+    .getLogs("lifecycle-output")
+    .map((entry) => entry.message)
+    .filter((message) => message === "initialized" || message === "initialization warning")
+    .sort();
+}
+
 afterEach(async () => {
   await Promise.all(
     temporaryDirectories.splice(0).map((directory) => rm(directory, { recursive: true })),
@@ -65,6 +147,122 @@ afterEach(async () => {
 });
 
 describe("PluginRuntime", () => {
+  it("frames stdout and stderr, normalizes CRLF, and flushes final fragments once", async () => {
+    const directory = await createPlugin(
+      "output",
+      `export default function contribute(plugin: unknown) { void plugin; return () => undefined; }`,
+    );
+    const child = createReloadChild("output", []);
+    const runtime = createTestRuntime({ spawnChild: () => child });
+    await runtime.startPlugin("output", directory);
+
+    child.stdout.write("first\r");
+    child.stdout.write("\n");
+    child.stderr.write("problem\nfinal stderr");
+    child.stdout.write("final stdout");
+    await runtime.stopPluginById("output");
+
+    expect(runtime.getLogs("output").map(({ stream, message }) => ({ stream, message }))).toEqual([
+      { stream: "stdout", message: "first" },
+      { stream: "stderr", message: "problem" },
+      { stream: "stderr", message: "final stderr" },
+      { stream: "stdout", message: "final stdout" },
+    ]);
+    expect(runtime.getLogs("output").map((entry) => entry.sequence)).toEqual([1, 2, 3, 4]);
+  });
+
+  it("writes plugin output through the daemon logger with structured identity fields", async () => {
+    const directory = await createPlugin(
+      "tagged-output",
+      `export default function contribute(plugin: unknown) { void plugin; return () => undefined; }`,
+    );
+    const child = createReloadChild("tagged-output", []);
+    const records: Array<Record<string, unknown>> = [];
+    const logger = pino(
+      { level: "info" },
+      { write: (line: string) => records.push(JSON.parse(line) as Record<string, unknown>) },
+    );
+    const runtime = createTestRuntime({ spawnChild: () => child }, logger);
+    await runtime.startPlugin("tagged-output", directory);
+
+    child.stderr.write("connection failed\n");
+
+    expect(records.find((record) => record.msg === "Plugin output")).toMatchObject({
+      module: "plugins",
+      pluginId: "tagged-output",
+      sequence: 1,
+      stream: "stderr",
+      message: "connection failed",
+    });
+    expect(records.find((record) => record.msg === "Plugin output")?.timestamp).toEqual(
+      expect.any(String),
+    );
+    await runtime.stopAll();
+  });
+
+  it("bounds retained output by entry count, total bytes, and individual line bytes", async () => {
+    const directory = await createPlugin(
+      "noisy",
+      `export default function contribute(plugin: unknown) { void plugin; return () => undefined; }`,
+    );
+    const child = createReloadChild("noisy", []);
+    const runtime = createTestRuntime({ spawnChild: () => child });
+    await runtime.startPlugin("noisy", directory);
+
+    child.stdout.write(
+      `${Array.from({ length: 510 }, (_, index) => `line-${index}`).join("\n")}\n`,
+    );
+    expect(runtime.getLogs("noisy")).toHaveLength(500);
+    expect(runtime.getLogs("noisy")[0]?.message).toBe("line-10");
+
+    child.stdout.write(`${Array.from({ length: 20 }, () => "x".repeat(16 * 1024)).join("\n")}\n`);
+    expect(runtime.getLogs("noisy")).toHaveLength(16);
+
+    child.stderr.write("y".repeat(20 * 1024));
+    await runtime.stopPluginById("noisy");
+    const logs = runtime.getLogs("noisy");
+    expect(logs).toHaveLength(16);
+    expect(logs.at(-1)).toMatchObject({ stream: "stderr", message: "y".repeat(16 * 1024) });
+  });
+
+  it("captures output emitted during initialization and cleanup across reloads", async () => {
+    const directory = await createPlugin(
+      "lifecycle-output",
+      `export default function contribute(plugin: unknown) {
+  void plugin;
+  console.log("initialized");
+  console.error("initialization warning");
+  return () => process.stdout.write("cleanup fragment");
+}`,
+    );
+    const runtime = createTestRuntime();
+
+    await runtime.startPlugin("lifecycle-output", directory);
+    await expect
+      .poll(() => lifecycleMessages(runtime))
+      .toEqual(["initialization warning", "initialized"]);
+    await runtime.stopPluginById("lifecycle-output");
+    await runtime.startPlugin("lifecycle-output", directory);
+    await runtime.stopPluginById("lifecycle-output");
+
+    const logs = runtime.getLogs("lifecycle-output");
+    expect(
+      logs
+        .filter((entry) => entry.stream === "stdout")
+        .map((entry) => entry.message)
+        .filter((message) => message === "initialized" || message === "cleanup fragment"),
+    ).toEqual(["initialized", "cleanup fragment", "initialized", "cleanup fragment"]);
+    expect(
+      logs
+        .filter((entry) => entry.stream === "stderr")
+        .map((entry) => entry.message)
+        .filter((message) => message === "initialization warning"),
+    ).toEqual(["initialization warning", "initialization warning"]);
+    expect(logs.map((entry) => entry.sequence)).toEqual(
+      Array.from({ length: logs.length }, (_, index) => index + 1),
+    );
+  });
+
   it("waits for the old subprocess to exit before starting its replacement", async () => {
     const directory = await createPlugin(
       "reloadable",
@@ -72,7 +270,7 @@ describe("PluginRuntime", () => {
     );
     const events: string[] = [];
     const children = [createReloadChild("old", events), createReloadChild("new", events)];
-    const runtime = new PluginRuntime(pino({ level: "silent" }), {
+    const runtime = createTestRuntime({
       spawnChild: () => {
         const child = children.shift();
         if (!child) throw new Error("Unexpected extra child");
@@ -94,7 +292,7 @@ describe("PluginRuntime", () => {
       `export default function contribute(plugin: unknown) { void plugin; return () => undefined; }`,
     );
     const child = createReloadChild("pending", [], ["wait"]);
-    const runtime = new PluginRuntime(pino({ level: "silent" }), {
+    const runtime = createTestRuntime({
       spawnChild: () => child,
     });
     await runtime.startPlugin("pending", directory);
@@ -107,6 +305,65 @@ describe("PluginRuntime", () => {
     await rejection;
   });
 
+  it("waits for asynchronous plugin cleanup before stopping", async () => {
+    const cleanupFile = path.join(tmpdir(), `paseo-plugin-cleanup-${Date.now()}`);
+    const directory = await createPlugin(
+      "async-cleanup",
+      `import { writeFile } from "node:fs/promises";
+export default function contribute(plugin: unknown) {
+  void plugin;
+  return async () => {
+    await new Promise((resolve) => setTimeout(resolve, 20));
+    await writeFile(${JSON.stringify(cleanupFile)}, "cleaned");
+  };
+}`,
+    );
+    const runtime = createTestRuntime();
+    await runtime.startPlugin("async-cleanup", directory);
+
+    await runtime.stopPluginById("async-cleanup");
+
+    await expect(readFile(cleanupFile, "utf8")).resolves.toBe("cleaned");
+    await rm(cleanupFile, { force: true });
+  });
+
+  it("does not kill a healthy child while its graceful cleanup is still running", async () => {
+    vi.useFakeTimers();
+    try {
+      const directory = await createPlugin(
+        "held-cleanup",
+        `export default function contribute(plugin: unknown) { void plugin; return () => undefined; }`,
+      );
+      const events: string[] = [];
+      const child = createReloadChild("held-cleanup", events);
+      const originalSend = child.send.bind(child);
+      let releaseCleanup = () => undefined;
+      child.send = (message, callback) => {
+        if (message.type !== "shutdown") return originalSend(message, callback);
+        callback?.(null);
+        events.push("shutdown:held-cleanup");
+        releaseCleanup = () => {
+          child.connected = false;
+          child.kill();
+          child.killed = false;
+        };
+        return true;
+      };
+      const runtime = createTestRuntime({ spawnChild: () => child });
+      await runtime.startPlugin("held-cleanup", directory);
+
+      const stopping = runtime.stopPluginById("held-cleanup");
+      await vi.advanceTimersByTimeAsync(60_000);
+      expect(child.killed).toBe(false);
+
+      releaseCleanup();
+      await stopping;
+      expect(events).toContain("shutdown:held-cleanup");
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
   it("kills a plugin child that fails initialization", async () => {
     const directory = await createPlugin(
       "broken",
@@ -142,7 +399,7 @@ describe("PluginRuntime", () => {
         return this;
       },
     };
-    const runtime = new PluginRuntime(pino({ level: "silent" }), {
+    const runtime = createTestRuntime({
       spawnChild: () => child,
     });
 
@@ -157,7 +414,7 @@ describe("PluginRuntime", () => {
       "missing-cleanup",
       `export default function contribute(plugin: unknown) { void plugin; }`,
     );
-    const runtime = new PluginRuntime(pino({ level: "silent" }));
+    const runtime = createTestRuntime();
 
     await expect(runtime.startPlugin("missing-cleanup", directory)).rejects.toThrow(
       "must return a cleanup function",
@@ -171,7 +428,7 @@ describe("PluginRuntime", () => {
       path.dirname(fileURLToPath(import.meta.url)),
       "../../../../../plugin-examples/linear",
     );
-    const runtime = new PluginRuntime(pino({ level: "silent" }));
+    const runtime = createTestRuntime();
 
     await runtime.startPlugin("linear", directory);
 
@@ -221,7 +478,7 @@ export default function contribute(plugin: any) {
   return () => undefined;
 }`,
     );
-    const runtime = new PluginRuntime(pino({ level: "silent" }));
+    const runtime = createTestRuntime();
 
     await runtime.startPlugin("hello", directory);
 
@@ -255,7 +512,7 @@ export default function contribute(plugin: any) {
   return () => undefined;
 }`,
     );
-    const runtime = new PluginRuntime(pino({ level: "silent" }));
+    const runtime = createTestRuntime();
 
     await runtime.startPlugin("invalid-output", directory);
 
@@ -268,7 +525,7 @@ export default function contribute(plugin: any) {
       "actual",
       `export default function contribute(plugin: unknown) { void plugin; return () => undefined; }`,
     );
-    const runtime = new PluginRuntime(pino({ level: "silent" }));
+    const runtime = createTestRuntime();
 
     await runtime.startPlugin("configured", directory);
 
@@ -283,7 +540,7 @@ export default function contribute(plugin: any) {
     );
     const events: string[] = [];
     const child = createReloadChild("blocked", events);
-    const runtime = new PluginRuntime(pino({ level: "silent" }), { spawnChild: () => child });
+    const runtime = createTestRuntime({ spawnChild: () => child });
 
     await expect(runtime.startPlugin("blocked", directory, () => false)).rejects.toThrow(
       "Plugin start cancelled: blocked",
@@ -298,11 +555,13 @@ export default function contribute(plugin: any) {
       "crashing",
       `export default function contribute(plugin: unknown) {
   void plugin;
+  process.stdout.write("before crash");
   setTimeout(() => process.exit(17), 20);
   return () => undefined;
 }`,
     );
-    const runtime = new PluginRuntime(pino({ level: "silent" }));
+    const sessions = createTrackedSessionHost();
+    const runtime = createTestRuntime({ sessionHost: sessions.host });
     const crashed = new Promise<string>((resolve) => {
       runtime.subscribe((pluginId, error) => {
         if (pluginId === "crashing" && error) resolve(error);
@@ -311,7 +570,9 @@ export default function contribute(plugin: any) {
     await runtime.startPlugin("crashing", directory);
 
     await expect(crashed).resolves.toBe("Plugin process exited: crashing");
+    expect(sessions.active.size).toBe(0);
     expect(runtime.catalog()).toEqual([]);
+    expect(runtime.getLogs("crashing").map((entry) => entry.message)).toContain("before crash");
     await expect(runtime.invoke("crashing", "anything", {})).rejects.toThrow(
       "Plugin is not available",
     );

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

@@ -4,16 +4,27 @@ import path from "node:path";
 import { fileURLToPath } from "node:url";
 import { randomUUID } from "node:crypto";
 import type pino from "pino";
+import type { PluginLogEntry } from "@getpaseo/protocol/messages";
 import { compilePlugin } from "./compiler.js";
 import { readPluginManifest } from "./manifest.js";
 import type { PluginProcessMessage, PluginProcessRequest } from "./plugin-process-protocol.js";
+import { PluginSessionSocket } from "./session-socket.js";
 
 const ENTRY_FILENAME = "index.tsx";
 const REQUEST_TIMEOUT_MS = 30_000;
+const MAX_LOG_ENTRIES = 500;
+const MAX_LOG_BYTES = 256 * 1024;
+const MAX_LOG_LINE_BYTES = 16 * 1024;
+
+interface PluginOutputStream {
+  on(event: "data", listener: (chunk: Buffer | string) => void): this;
+}
 
 interface PluginChild {
   connected: boolean;
   killed: boolean;
+  stdout?: PluginOutputStream | null;
+  stderr?: PluginOutputStream | null;
   send(message: PluginProcessRequest, callback?: (error: Error | null) => void): boolean;
   kill(): boolean;
   disconnect(): void;
@@ -32,11 +43,95 @@ interface LoadedPlugin {
   clientBundle: string;
   methods: ReadonlySet<string>;
   child: PluginChild;
+  outputCapture: PluginOutputCapture;
   pending: Map<string, PendingInvocation>;
+  sessionSocket: PluginSessionSocket;
+  sessionClosed: Promise<void>;
 }
 
 interface PluginRuntimeDependencies {
   spawnChild?: () => PluginChild;
+  sessionHost?: PluginPaseoSessionHost;
+}
+
+interface PluginLogTail {
+  entries: PluginLogEntry[];
+  bytes: number;
+  nextSequence: number;
+}
+
+class PluginOutputCapture {
+  private readonly pending = new Map<PluginLogEntry["stream"], Buffer>([
+    ["stdout", Buffer.alloc(0)],
+    ["stderr", Buffer.alloc(0)],
+  ]);
+  private readonly overflowed = new Set<PluginLogEntry["stream"]>();
+  private readonly lastActivity = new Map<PluginLogEntry["stream"], number>();
+  private activitySequence = 0;
+  private flushed = false;
+
+  constructor(
+    child: PluginChild,
+    private readonly emit: (stream: PluginLogEntry["stream"], message: string) => void,
+  ) {
+    child.stdout?.on("data", (chunk) => this.write("stdout", chunk));
+    child.stderr?.on("data", (chunk) => this.write("stderr", chunk));
+    child.on("close", () => this.flush());
+  }
+
+  private write(stream: PluginLogEntry["stream"], chunk: Buffer | string): void {
+    if (this.flushed) return;
+    const data = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk);
+    let offset = 0;
+    while (offset < data.length) {
+      const newline = data.indexOf(0x0a, offset);
+      const end = newline === -1 ? data.length : newline;
+      this.append(stream, data.subarray(offset, end));
+      if (newline === -1) return;
+      this.emitLine(stream);
+      offset = newline + 1;
+    }
+  }
+
+  private append(stream: PluginLogEntry["stream"], chunk: Buffer): void {
+    if (chunk.length > 0) this.lastActivity.set(stream, ++this.activitySequence);
+    const current = this.pending.get(stream) ?? Buffer.alloc(0);
+    const remaining = MAX_LOG_LINE_BYTES - current.length;
+    if (chunk.length > remaining) this.overflowed.add(stream);
+    if (remaining <= 0) return;
+    this.pending.set(stream, Buffer.concat([current, chunk.subarray(0, remaining)]));
+  }
+
+  private emitLine(stream: PluginLogEntry["stream"]): void {
+    let line = this.pending.get(stream) ?? Buffer.alloc(0);
+    if (!this.overflowed.has(stream) && line.at(-1) === 0x0d) line = line.subarray(0, -1);
+    this.emit(stream, line.toString("utf8"));
+    this.pending.set(stream, Buffer.alloc(0));
+    this.overflowed.delete(stream);
+    this.lastActivity.delete(stream);
+  }
+
+  private flush(): void {
+    if (this.flushed) return;
+    this.flushed = true;
+    const pendingStreams = (["stdout", "stderr"] as const)
+      .filter(
+        (stream) => (this.pending.get(stream)?.length ?? 0) > 0 || this.overflowed.has(stream),
+      )
+      .sort(
+        (left, right) => (this.lastActivity.get(left) ?? 0) - (this.lastActivity.get(right) ?? 0),
+      );
+    for (const stream of pendingStreams) {
+      this.emitLine(stream);
+    }
+  }
+}
+
+export interface PluginPaseoSessionHost {
+  attachPluginSocket(
+    pluginId: string,
+    socket: PluginSessionSocket,
+  ): Promise<{ closed: Promise<void> }>;
 }
 
 function resolveWorkerUrl(): URL {
@@ -65,7 +160,7 @@ function spawnPluginChild(): PluginChild {
   return fork(fileURLToPath(resolveWorkerUrl()), [], {
     execArgv: resolveWorkerExecArgv(),
     serialization: "advanced",
-    stdio: ["ignore", "ignore", "inherit", "ipc"],
+    stdio: ["ignore", "pipe", "pipe", "ipc"],
   }) as PluginChild;
 }
 
@@ -93,13 +188,22 @@ async function requireRegularFile(filePath: string, label: string): Promise<void
 
 export class PluginRuntime {
   private readonly plugins = new Map<string, LoadedPlugin>();
+  private readonly logTails = new Map<string, PluginLogTail>();
   private readonly logger: pino.Logger;
   private readonly spawnChild: () => PluginChild;
+  private sessionHost: PluginPaseoSessionHost | null;
   private readonly listeners = new Set<(pluginId: string, error?: string) => void>();
 
   constructor(logger: pino.Logger, dependencies: PluginRuntimeDependencies = {}) {
     this.logger = logger.child({ module: "plugins" });
     this.spawnChild = dependencies.spawnChild ?? spawnPluginChild;
+    this.sessionHost = dependencies.sessionHost ?? null;
+  }
+
+  bindPaseoSessionHost(sessionHost: PluginPaseoSessionHost): void {
+    if (this.plugins.size > 0)
+      throw new Error("Cannot replace the plugin session host while running");
+    this.sessionHost = sessionHost;
   }
 
   subscribe(listener: (pluginId: string, error?: string) => void): () => void {
@@ -136,6 +240,21 @@ export class PluginRuntime {
       .sort((left, right) => left.id.localeCompare(right.id));
   }
 
+  getLogs(pluginId: string): PluginLogEntry[] {
+    return (
+      this.logTails.get(pluginId)?.entries.map((entry) => ({
+        sequence: entry.sequence,
+        timestamp: entry.timestamp,
+        stream: entry.stream,
+        message: entry.message,
+      })) ?? []
+    );
+  }
+
+  clearLogs(pluginId: string): void {
+    this.logTails.delete(pluginId);
+  }
+
   async invoke(pluginId: string, method: string, input: unknown): Promise<unknown> {
     const loaded = this.plugins.get(pluginId);
     if (!loaded) throw new Error(`Plugin is not available: ${pluginId}`);
@@ -174,8 +293,21 @@ export class PluginRuntime {
     const entryPath = path.join(directory, ENTRY_FILENAME);
     await requireRegularFile(entryPath, "Plugin entry point");
     const bundles = await compilePlugin(entryPath);
+    const sessionHost = this.sessionHost;
+    if (!sessionHost) throw new Error("Plugin Paseo session host is not attached");
     const child = this.spawnChild();
+    const outputCapture = new PluginOutputCapture(child, (stream, message) => {
+      this.appendLog(pluginId, stream, message);
+    });
+    const sessionSocket = new PluginSessionSocket(child);
     const pending = new Map<string, PendingInvocation>();
+    const sessionAttachment = await sessionHost
+      .attachPluginSocket(pluginId, sessionSocket)
+      .catch((error) => {
+        terminatePluginChild(child);
+        throw error;
+      });
+    let loaded: LoadedPlugin | null = null;
     let methods: string[];
     try {
       methods = await new Promise<string[]>((resolve, reject) => {
@@ -191,31 +323,51 @@ export class PluginRuntime {
           reject(error);
         };
         child.on("message", (message) => {
-          if (message.type === "ready") {
+          if (message.type === "paseo_frame") {
+            sessionSocket.receive(message.data, message.isBinary);
+          } else if (message.type === "paseo_close") {
+            sessionSocket.peerClosed();
+          } else if (message.type === "ready") {
             if (settled) return;
             settled = true;
             clearTimeout(timeout);
             resolve(message.methods);
           } else if (message.type === "fatal") {
             fail(new Error(message.error));
+          } else if (loaded) {
+            this.handleChildMessage(loaded, message);
           }
         });
-        child.on("close", () => fail(new Error(`Plugin ${pluginId} exited during initialization`)));
-        void send(child, { type: "initialize", bundle: bundles.serverBundle }).catch(fail);
+        child.on("close", () => {
+          sessionSocket.peerClosed();
+          if (!loaded) {
+            fail(new Error(`Plugin ${pluginId} exited during initialization`));
+            return;
+          }
+          void this.handleChildClose(loaded);
+        });
+        void send(child, {
+          type: "initialize",
+          pluginId,
+          bundle: bundles.serverBundle,
+        }).catch(fail);
       });
     } catch (error) {
+      sessionSocket.close();
+      await sessionAttachment.closed;
       terminatePluginChild(child);
       throw error;
     }
-    const loaded: LoadedPlugin = {
+    loaded = {
       id: pluginId,
       clientBundle: bundles.clientBundle,
       methods: new Set(methods),
       child,
+      outputCapture,
       pending,
+      sessionSocket,
+      sessionClosed: sessionAttachment.closed,
     };
-    child.on("message", (message) => this.handleChildMessage(loaded, message));
-    child.on("close", () => this.handleChildClose(loaded));
     this.logger.info({ pluginId, methods }, "Loaded plugin");
     return loaded;
   }
@@ -230,29 +382,34 @@ export class PluginRuntime {
     else pending.reject(new Error(message.error));
   }
 
-  private handleChildClose(loaded: LoadedPlugin): void {
-    if (this.plugins.get(loaded.id) === loaded) {
+  private async handleChildClose(loaded: LoadedPlugin): Promise<void> {
+    loaded.sessionSocket.peerClosed();
+    const wasPublished = this.plugins.get(loaded.id) === loaded;
+    if (wasPublished) {
       this.plugins.delete(loaded.id);
-      this.notify(loaded.id, `Plugin process exited: ${loaded.id}`);
     }
     this.rejectPending(loaded, `Plugin process exited: ${loaded.id}`);
+    await loaded.sessionClosed;
+    if (wasPublished) this.notify(loaded.id, `Plugin process exited: ${loaded.id}`);
   }
 
   private async stopPlugin(loaded: LoadedPlugin): Promise<void> {
-    if (loaded.child.killed) return;
-    let didClose = false;
+    if (loaded.child.killed) {
+      loaded.sessionSocket.peerClosed();
+      await loaded.sessionClosed;
+      return;
+    }
     const closed = new Promise<void>((resolve) =>
       loaded.child.on("close", () => {
-        didClose = true;
         resolve();
       }),
     );
     if (loaded.child.connected) {
       await send(loaded.child, { type: "shutdown" }).catch(() => undefined);
     }
-    await Promise.race([closed, new Promise<void>((resolve) => setTimeout(resolve, 1_000))]);
-    if (!didClose) terminatePluginChild(loaded.child);
     await closed;
+    loaded.sessionSocket.peerClosed();
+    await loaded.sessionClosed;
   }
 
   private rejectPending(loaded: LoadedPlugin, message: string): void {
@@ -263,6 +420,28 @@ export class PluginRuntime {
     loaded.pending.clear();
   }
 
+  private appendLog(pluginId: string, stream: PluginLogEntry["stream"], message: string): void {
+    let tail = this.logTails.get(pluginId);
+    if (!tail) {
+      tail = { entries: [], bytes: 0, nextSequence: 1 };
+      this.logTails.set(pluginId, tail);
+    }
+    const entry: PluginLogEntry = {
+      sequence: tail.nextSequence++,
+      timestamp: new Date().toISOString(),
+      stream,
+      message,
+    };
+    tail.entries.push(entry);
+    tail.bytes += Buffer.byteLength(message);
+    while (tail.entries.length > MAX_LOG_ENTRIES || tail.bytes > MAX_LOG_BYTES) {
+      const removed = tail.entries.shift();
+      if (!removed) break;
+      tail.bytes -= Buffer.byteLength(removed.message);
+    }
+    this.logger.info({ pluginId, ...entry }, "Plugin output");
+  }
+
   private notify(pluginId: string, error?: string): void {
     for (const listener of this.listeners) listener(pluginId, error);
   }

packages/server/src/server/plugins/session-socket.posix.test.ts

@@ -0,0 +1,32 @@
+import { fork } from "node:child_process";
+import { fileURLToPath } from "node:url";
+import { describe, expect, it } from "vitest";
+import type { PluginProcessMessage } from "./plugin-process-protocol.js";
+import { PluginSessionSocket } from "./session-socket.js";
+
+describe("PluginSessionSocket child-process transport", () => {
+  it("round-trips ordered text and binary frames through a real child process", async () => {
+    const child = fork(
+      fileURLToPath(new URL("./test-fixtures/plugin-session-echo.cjs", import.meta.url)),
+      [],
+      { serialization: "advanced", stdio: ["ignore", "ignore", "ignore", "ipc"] },
+    );
+    const socket = new PluginSessionSocket(child);
+    const received: Array<string | Uint8Array> = [];
+    socket.on("message", (data) => received.push(data as string | Uint8Array));
+    child.on("message", (message: PluginProcessMessage) => {
+      if (message.type === "paseo_frame") socket.receive(message.data, message.isBinary);
+    });
+
+    try {
+      socket.send("first");
+      socket.send(new Uint8Array([1, 2, 3]));
+      socket.send("last");
+      await expect.poll(() => received.length).toBe(3);
+
+      expect(received).toEqual(["first", new Uint8Array([1, 2, 3]), "last"]);
+    } finally {
+      child.kill();
+    }
+  });
+});

packages/server/src/server/plugins/session-socket.test.ts

@@ -0,0 +1,45 @@
+import { describe, expect, it } from "vitest";
+import type { PluginProcessRequest } from "./plugin-process-protocol.js";
+import { PluginSessionSocket } from "./session-socket.js";
+
+describe("PluginSessionSocket", () => {
+  it("preserves text and binary frame order across IPC", () => {
+    const sent: PluginProcessRequest[] = [];
+    const socket = new PluginSessionSocket({
+      send(message, callback) {
+        sent.push(message);
+        callback?.(null);
+        return true;
+      },
+    });
+
+    socket.send("first");
+    socket.send(new Uint8Array([1, 2, 3]));
+    socket.send("last");
+
+    expect(sent).toEqual([
+      { type: "paseo_frame", data: "first", isBinary: false },
+      { type: "paseo_frame", data: new Uint8Array([1, 2, 3]), isBinary: true },
+      { type: "paseo_frame", data: "last", isBinary: false },
+    ]);
+  });
+
+  it("reports bytes queued in child-process IPC for terminal backpressure", () => {
+    const callbacks: Array<(error: Error | null) => void> = [];
+    const socket = new PluginSessionSocket({
+      send(_message, callback) {
+        if (callback) callbacks.push(callback);
+        return true;
+      },
+    });
+
+    socket.send("hello");
+    socket.send(new Uint8Array([1, 2, 3]));
+    expect(socket.bufferedAmount).toBe(8);
+
+    callbacks[0]?.(null);
+    expect(socket.bufferedAmount).toBe(3);
+    callbacks[1]?.(null);
+    expect(socket.bufferedAmount).toBe(0);
+  });
+});

packages/server/src/server/plugins/session-socket.ts

@@ -0,0 +1,93 @@
+import { Buffer } from "node:buffer";
+import type { PluginProcessRequest } from "./plugin-process-protocol.js";
+
+interface PluginProcessSender {
+  send(message: PluginProcessRequest, callback?: (error: Error | null) => void): boolean;
+}
+
+type SocketEvent = "message" | "close" | "error";
+type SocketListener = (...args: unknown[]) => void;
+
+function normalizeBinaryFrame(data: Uint8Array | ArrayBuffer): Uint8Array {
+  if (data instanceof Uint8Array) {
+    return new Uint8Array(data.buffer, data.byteOffset, data.byteLength);
+  }
+  return new Uint8Array(data);
+}
+
+function frameByteLength(data: string | Uint8Array): number {
+  return typeof data === "string" ? Buffer.byteLength(data) : data.byteLength;
+}
+
+export class PluginSessionSocket {
+  readyState = 1;
+  bufferedAmount = 0;
+  private readonly listeners = new Map<SocketEvent, Set<SocketListener>>();
+  private readonly onceListeners = new Map<SocketEvent, Set<SocketListener>>();
+
+  constructor(private readonly child: PluginProcessSender) {}
+
+  send(data: string | Uint8Array | ArrayBuffer, callback?: (error?: Error) => void): void {
+    if (this.readyState !== 1) {
+      callback?.(new Error("Plugin session socket is closed"));
+      return;
+    }
+    const frame = typeof data === "string" ? data : normalizeBinaryFrame(data);
+    const byteLength = frameByteLength(frame);
+    this.bufferedAmount += byteLength;
+    try {
+      this.child.send(
+        { type: "paseo_frame", data: frame, isBinary: typeof frame !== "string" },
+        (error) => {
+          this.bufferedAmount = Math.max(0, this.bufferedAmount - byteLength);
+          callback?.(error ?? undefined);
+        },
+      );
+    } catch (error) {
+      this.bufferedAmount = Math.max(0, this.bufferedAmount - byteLength);
+      callback?.(error instanceof Error ? error : new Error(String(error)));
+    }
+  }
+
+  receive(data: string | Uint8Array, isBinary: boolean): void {
+    if (this.readyState !== 1) return;
+    this.emit("message", data, isBinary);
+  }
+
+  close(code?: number, reason?: string): void {
+    if (this.readyState === 3) return;
+    this.readyState = 3;
+    try {
+      this.child.send({ type: "paseo_close" });
+    } catch {
+      // The child has already gone; local session cleanup still has to run.
+    }
+    this.emit("close", code, reason);
+  }
+
+  peerClosed(): void {
+    if (this.readyState === 3) return;
+    this.readyState = 3;
+    this.emit("close");
+  }
+
+  on(event: SocketEvent, listener: SocketListener): void {
+    const listeners = this.listeners.get(event) ?? new Set<SocketListener>();
+    listeners.add(listener);
+    this.listeners.set(event, listeners);
+  }
+
+  once(event: "close" | "error", listener: SocketListener): void {
+    const listeners = this.onceListeners.get(event) ?? new Set<SocketListener>();
+    listeners.add(listener);
+    this.onceListeners.set(event, listeners);
+  }
+
+  private emit(event: SocketEvent, ...args: unknown[]): void {
+    for (const listener of this.listeners.get(event) ?? []) listener(...args);
+    const once = this.onceListeners.get(event);
+    if (!once) return;
+    this.onceListeners.delete(event);
+    for (const listener of once) listener(...args);
+  }
+}

packages/server/src/server/plugins/test-fixtures/plugin-session-echo.cjs

@@ -0,0 +1,4 @@
+process.on("message", (message) => {
+  if (message?.type !== "paseo_frame") return;
+  process.send?.(message);
+});

packages/server/src/server/session.test.ts

@@ -426,9 +426,12 @@ function createSessionForTest(options: SessionForTestOptions = {}): Session {
   return new Session(sessionOptions);
 }
 
-test("routes plugin management requests and catalog notifications", async () => {
+test("routes plugin requests and releases its owned catalog subscription on cleanup", async () => {
   const messages: SessionOutboundMessage[] = [];
   const listeners = new Set<(pluginId: string) => void>();
+  const releasePluginSubscription = vi.fn((listener: (pluginId: string) => void) => {
+    listeners.delete(listener);
+  });
   const plugin = {
     id: "example",
     path: "/plugins/example",
@@ -437,6 +440,14 @@ test("routes plugin management requests and catalog notifications", async () =>
   };
   const pluginRuntime: NonNullable<SessionOptions["pluginRuntime"]> = {
     listPlugins: () => [plugin],
+    getLogs: () => [
+      {
+        sequence: 1,
+        timestamp: "2026-08-16T12:00:00.000Z",
+        stream: "stdout",
+        message: "ready",
+      },
+    ],
     installDirectory: async () => plugin,
     inspectDirectory: async () => ({ id: "example" }),
     reloadPlugin: async () => plugin,
@@ -445,7 +456,7 @@ test("routes plugin management requests and catalog notifications", async () =>
     removePlugin: async () => undefined,
     subscribe: (listener) => {
       listeners.add(listener);
-      return () => listeners.delete(listener);
+      return () => releasePluginSubscription(listener);
     },
     catalog: () => [{ id: "example", clientBundle: "bundle" }],
     invokePluginRpc: async () => ({ ok: true }),
@@ -453,6 +464,11 @@ test("routes plugin management requests and catalog notifications", async () =>
   const session = createSessionForTest({ messages, pluginRuntime });
 
   await session.handleMessage({ type: "plugin.list.request", requestId: "list" });
+  await session.handleMessage({
+    type: "plugin.logs.get.request",
+    requestId: "logs",
+    pluginId: "example",
+  });
   await session.handleMessage({
     type: "plugin.directory.install.request",
     requestId: "install",
@@ -477,6 +493,7 @@ test("routes plugin management requests and catalog notifications", async () =>
 
   expect(messages.map((message) => message.type)).toEqual([
     "plugin.list.response",
+    "plugin.logs.get.response",
     "plugin.directory.install.response",
     "plugin.reload.response",
     "plugin.disable.response",
@@ -489,6 +506,7 @@ test("routes plugin management requests and catalog notifications", async () =>
   });
   await session.cleanup();
   expect(listeners.size).toBe(0);
+  expect(releasePluginSubscription).toHaveBeenCalledOnce();
 });
 
 describe("session authorization scopes", () => {

packages/server/src/server/session.ts

@@ -462,6 +462,7 @@ export interface SessionOptions {
   daemonConfigStore: DaemonConfigStore;
   pluginRuntime?: {
     listPlugins(): import("@getpaseo/protocol/messages").PluginListItem[];
+    getLogs(pluginId: string): import("@getpaseo/protocol/messages").PluginLogEntry[];
     installDirectory(input: {
       path: string;
       id?: string;
@@ -1898,6 +1899,18 @@ export class Session {
       });
       return undefined;
     }
+    if (msg.type === "plugin.logs.get.request") {
+      if (!this.pluginRuntime) throw new Error("Plugin service is unavailable");
+      this.emit({
+        type: "plugin.logs.get.response",
+        payload: {
+          requestId: msg.requestId,
+          pluginId: msg.pluginId,
+          entries: this.pluginRuntime.getLogs(msg.pluginId),
+        },
+      });
+      return undefined;
+    }
     if (msg.type === "plugin.reload.request") {
       if (!this.pluginRuntime) throw new Error("Plugin service is unavailable");
       return this.pluginRuntime.reloadPlugin(msg.pluginId).then((plugin) => {

packages/server/src/server/test-utils/paseo-daemon.ts

@@ -48,6 +48,8 @@ interface TestPaseoDaemonOptions {
   trustedProxies?: PaseoDaemonConfig["trustedProxies"];
   agentProfiles?: AgentProfile[];
   autoArchiveAfterMerge?: boolean;
+  pluginsEnabled?: PaseoDaemonConfig["pluginsEnabled"];
+  plugins?: PaseoDaemonConfig["plugins"];
 }
 
 export interface TestPaseoDaemon {
@@ -199,6 +201,8 @@ async function prepareTestDaemonConfig(
     downloadTokenTtlMs: options.downloadTokenTtlMs,
     agentProfiles: options.agentProfiles,
     autoArchiveAfterMerge: options.autoArchiveAfterMerge,
+    pluginsEnabled: options.pluginsEnabled,
+    plugins: options.plugins,
   };
   return { config, paseoHomeRoot, paseoHome, staticDir };
 }

packages/server/src/server/websocket-server.relay-reconnect.test.ts

@@ -220,6 +220,7 @@ function createWorkspaceAutoNameStub(): WorkspaceAutoName {
 function createServer(options?: {
   speechReadiness?: SpeechReadinessSnapshot | null;
   logger?: ReturnType<typeof createLogger>;
+  startPaused?: boolean;
 }) {
   const speechReadiness = options?.speechReadiness ?? null;
   const daemonConfigStore = {
@@ -248,7 +249,7 @@ function createServer(options?: {
     "/tmp/paseo-test",
     createStub<DaemonConfigStore>(daemonConfigStore),
     null,
-    { allowedOrigins: new Set() },
+    { allowedOrigins: new Set(), startPaused: options?.startPaused },
     createWorkspaceAutoNameStub(),
     undefined,
     speechReadiness
@@ -503,6 +504,39 @@ describe("relay external socket reconnect behavior", () => {
     await server.close();
   });
 
+  test("gives every plugin socket an exclusively owned session and cleans it immediately", async () => {
+    const server = createServer();
+    const firstSocket = new MockSocket();
+    const firstAttachment = await server.attachPluginSocket("exclusive", firstSocket);
+    firstSocket.emit("message", JSON.stringify(createHelloMessage("plugin:exclusive")));
+
+    const secondSocket = new MockSocket();
+    const secondAttachment = await server.attachPluginSocket("exclusive", secondSocket);
+    secondSocket.emit("message", JSON.stringify(createHelloMessage("plugin:exclusive")));
+
+    expect(sessionMock.instances).toHaveLength(2);
+    firstSocket.emit("close", 1000, "plugin stopped");
+    await firstAttachment.closed;
+    expect(sessionMock.instances[0]?.cleanup).toHaveBeenCalledOnce();
+    expect(sessionMock.instances[1]?.cleanup).not.toHaveBeenCalled();
+
+    secondSocket.emit("close", 1000, "plugin stopped");
+    await secondAttachment.closed;
+    expect(sessionMock.instances[1]?.cleanup).toHaveBeenCalledOnce();
+    await server.close();
+  });
+
+  test("rejects ordinary sockets that claim the reserved plugin client id", async () => {
+    const server = createServer();
+    const socket = new MockSocket();
+    await server.attachExternalSocket(socket, { transport: "relay" });
+    socket.emit("message", JSON.stringify(createHelloMessage("plugin:not-a-plugin")));
+
+    expect(socket.readyState).toBe(3);
+    expect(sessionMock.instances).toHaveLength(0);
+    await server.close();
+  });
+
   test("passes hello capabilities through to the created session", async () => {
     const server = createServer();
     const socket = new MockSocket();
@@ -555,6 +589,27 @@ describe("relay external socket reconnect behavior", () => {
     }
   });
 
+  test("accepts plugin startup sessions while application sessions remain paused", async () => {
+    const server = createServer({ startPaused: true });
+    const applicationSocket = new MockSocket();
+    await server.attachExternalSocket(applicationSocket, { transport: "relay" });
+    expect(applicationSocket.readyState).toBe(3);
+
+    const pluginSocket = new MockSocket();
+    const attachment = await server.attachPluginSocket("startup", pluginSocket);
+    pluginSocket.emit("message", JSON.stringify(createHelloMessage("plugin:startup")));
+    expect(sessionMock.instances).toHaveLength(1);
+
+    server.beginAcceptingConnections();
+    const readySocket = new MockSocket();
+    await attachRelayAndHello({ server, socket: readySocket, clientId: "ready-client" });
+    expect(sessionMock.instances).toHaveLength(2);
+
+    pluginSocket.emit("close", 1000, "done");
+    await attachment.closed;
+    await server.close();
+  });
+
   test("closes pending connection when hello timeout elapses", async () => {
     const server = createServer();
 
@@ -923,6 +978,7 @@ describe("relay external socket reconnect behavior", () => {
     expect(serverInfo.features?.stableProjectIdentity).toBe(true);
     expect(serverInfo.features?.canonicalSubmittedPrompts).toBe(true);
     expect(serverInfo.features?.providersSnapshotCwd).toBe(true);
+    expect(serverInfo.features?.pluginLogs).toBe(true);
     expect(serverInfo.features?.["terminal-input-mode-replay"]).toBe(true);
     expect(serverInfo.features?.["terminal-size-ownership"]).toBe(true);
     expect(serverInfo.features?.agentTurnIdentity).toBeUndefined();

packages/server/src/server/websocket-server.ts

@@ -144,6 +144,7 @@ interface WebSocketServerConfig {
   getHostnames?: () => HostnamesConfig | undefined;
   daemonStatusRpc?: boolean;
   relayConfig?: boolean;
+  startPaused?: boolean;
 }
 
 type WebSocketRuntimeMetrics = SessionRuntimeMetrics & CheckoutDiffMetrics;
@@ -426,7 +427,7 @@ export interface WebSocketLike {
   once: (event: "close" | "error", listener: (...args: unknown[]) => void) => void;
 }
 
-interface TrustedSessionConnection {
+interface TrustedSessionConnectionBase {
   kind: "trusted";
   session: Session;
   clientId: string;
@@ -434,9 +435,20 @@ interface TrustedSessionConnection {
   clientCapabilities: Record<string, unknown> | null;
   connectionLogger: pino.Logger;
   sockets: Set<WebSocketLike>;
+}
+
+interface ReconnectableSessionConnection extends TrustedSessionConnectionBase {
+  lifecycle: "reconnectable";
   externalDisconnectCleanupTimeout: ReturnType<typeof setTimeout> | null;
 }
 
+interface PluginSessionConnection extends TrustedSessionConnectionBase {
+  lifecycle: "ephemeral-plugin";
+  pluginId: string;
+}
+
+type TrustedSessionConnection = ReconnectableSessionConnection | PluginSessionConnection;
+
 interface HubConnection {
   kind: "hub";
   session: Session;
@@ -529,7 +541,9 @@ export class VoiceAssistantWebSocketServer {
   private readonly pendingConnections: Map<WebSocketLike, PendingConnection> = new Map();
   private readonly sessions: Map<WebSocketLike, SessionConnection> = new Map();
   private readonly socketIdentities: Map<WebSocketLike, WebSocketConnectionIdentity> = new Map();
-  private readonly externalSessionsByKey: Map<string, TrustedSessionConnection> = new Map();
+  private readonly externalSessionsByKey: Map<string, ReconnectableSessionConnection> = new Map();
+  private readonly pluginSocketIds = new WeakMap<WebSocketLike, string>();
+  private readonly pluginSocketCleanup = new WeakMap<WebSocketLike, () => void>();
   private readonly serverId: string;
   private readonly daemonVersion: string;
   private readonly daemonRuntimeConfig: DaemonRuntimeConfig | undefined;
@@ -583,7 +597,7 @@ export class VoiceAssistantWebSocketServer {
   private readonly browserToolsBroker: BrowserToolsBroker | null;
   private readonly hubRelationships: HubRelationshipManagement | null;
   private readonly browserToolsRegistrations = new Map<string, BrowserToolsRegistration>();
-  private acceptingConnections = true;
+  private connectionLifecycle: "starting" | "accepting" | "stopping" = "accepting";
   private readonly advertiseDaemonStatusRpc: boolean;
   private readonly advertiseRelayConfig: boolean;
   private readonly directorySync = new DirectorySyncService();
@@ -638,6 +652,7 @@ export class VoiceAssistantWebSocketServer {
     this.workspaceSetupRuntime = workspaceSetupRuntime;
     this.advertiseDaemonStatusRpc = wsConfig.daemonStatusRpc !== false;
     this.advertiseRelayConfig = wsConfig.relayConfig !== false;
+    this.connectionLifecycle = wsConfig.startPaused === true ? "starting" : "accepting";
     this.serverId = serverId;
     if (typeof daemonVersion !== "string" || daemonVersion.trim().length === 0) {
       throw new MissingDaemonVersionError();
@@ -849,8 +864,8 @@ export class VoiceAssistantWebSocketServer {
     hostnames: HostnamesConfig | undefined,
     callback: (res: boolean, code?: number, message?: string) => void,
   ): void {
-    if (!this.acceptingConnections) {
-      callback(false, 503, "Server shutting down");
+    if (this.connectionLifecycle !== "accepting") {
+      callback(false, 503, "Server not ready");
       return;
     }
 
@@ -951,6 +966,29 @@ export class VoiceAssistantWebSocketServer {
     await this.attachSocket(ws, undefined, metadata);
   }
 
+  public async attachPluginSocket(
+    pluginId: string,
+    ws: WebSocketLike,
+  ): Promise<{ closed: Promise<void> }> {
+    if (this.connectionLifecycle === "stopping") {
+      throw new Error(`Cannot attach plugin session while shutting down: ${pluginId}`);
+    }
+    let resolve: () => void = () => undefined;
+    const closed = new Promise<void>((finish) => {
+      resolve = finish;
+    });
+    this.pluginSocketIds.set(ws, pluginId);
+    this.pluginSocketCleanup.set(ws, resolve);
+    try {
+      await this.attachSocket(ws, undefined, undefined, true);
+    } catch (error) {
+      this.pluginSocketIds.delete(ws);
+      this.finishPluginSocketCleanup(ws);
+      throw error;
+    }
+    return { closed };
+  }
+
   public async attachHubSocket(
     ws: WebSocketLike,
     options: {
@@ -959,7 +997,7 @@ export class VoiceAssistantWebSocketServer {
       agents: HubExecutionAgents;
     },
   ): Promise<void> {
-    if (!this.acceptingConnections) {
+    if (this.connectionLifecycle !== "accepting") {
       ws.close(WS_CLOSE_SERVER_SHUTDOWN, "Server shutting down");
       return;
     }
@@ -990,7 +1028,13 @@ export class VoiceAssistantWebSocketServer {
   }
 
   public prepareForShutdown(): void {
-    this.acceptingConnections = false;
+    this.connectionLifecycle = "stopping";
+  }
+
+  public beginAcceptingConnections(): void {
+    if (this.connectionLifecycle === "starting") {
+      this.connectionLifecycle = "accepting";
+    }
   }
 
   public async close(): Promise<void> {
@@ -1029,7 +1073,11 @@ export class VoiceAssistantWebSocketServer {
 
     const cleanupPromises: Promise<void>[] = [];
     for (const connection of uniqueConnections) {
-      if (connection.kind === "trusted" && connection.externalDisconnectCleanupTimeout) {
+      if (
+        connection.kind === "trusted" &&
+        connection.lifecycle === "reconnectable" &&
+        connection.externalDisconnectCleanupTimeout
+      ) {
         clearTimeout(connection.externalDisconnectCleanupTimeout);
         connection.externalDisconnectCleanupTimeout = null;
       }
@@ -1213,8 +1261,12 @@ export class VoiceAssistantWebSocketServer {
     ws: WebSocketLike,
     request?: unknown,
     metadata?: ExternalSocketMetadata,
+    allowDuringStartup = false,
   ): Promise<void> {
-    if (!this.acceptingConnections) {
+    if (
+      this.connectionLifecycle === "stopping" ||
+      (this.connectionLifecycle === "starting" && !allowDuringStartup)
+    ) {
       try {
         ws.close(WS_CLOSE_SERVER_SHUTDOWN, "Server shutting down");
       } catch {
@@ -1271,8 +1323,9 @@ export class VoiceAssistantWebSocketServer {
     appVersion: string | null;
     clientCapabilities: Record<string, unknown> | null;
     connectionLogger: pino.Logger;
+    lifecycle: { kind: "reconnectable" } | { kind: "ephemeral-plugin"; pluginId: string };
   }): TrustedSessionConnection {
-    const { ws, clientId, appVersion, clientCapabilities, connectionLogger } = params;
+    const { ws, clientId, appVersion, clientCapabilities, connectionLogger, lifecycle } = params;
     let connection: TrustedSessionConnection | null = null;
 
     const session = this.createSocketSession({
@@ -1327,7 +1380,7 @@ export class VoiceAssistantWebSocketServer {
       hubRelationships: this.hubRelationships ?? undefined,
     });
 
-    connection = {
+    const base: TrustedSessionConnectionBase = {
       kind: "trusted",
       session,
       clientId,
@@ -1335,8 +1388,11 @@ export class VoiceAssistantWebSocketServer {
       clientCapabilities,
       connectionLogger,
       sockets: new Set([ws]),
-      externalDisconnectCleanupTimeout: null,
     };
+    connection =
+      lifecycle.kind === "ephemeral-plugin"
+        ? { ...base, lifecycle: "ephemeral-plugin", pluginId: lifecycle.pluginId }
+        : { ...base, lifecycle: "reconnectable", externalDisconnectCleanupTimeout: null };
     session.updateClientCapabilities(clientCapabilities, ws);
     return connection;
   }
@@ -1477,48 +1533,26 @@ export class VoiceAssistantWebSocketServer {
       return;
     }
 
+    const pluginId = this.pluginSocketIds.get(ws);
+    const expectedPluginClientId = pluginId ? `plugin:${pluginId}` : null;
+    if (
+      (expectedPluginClientId !== null && clientId !== expectedPluginClientId) ||
+      (expectedPluginClientId === null && clientId.startsWith("plugin:"))
+    ) {
+      this.clearPendingConnection(ws);
+      pending.connectionLogger.warn({ clientId }, "Rejected reserved plugin clientId");
+      ws.close(WS_CLOSE_INVALID_HELLO, "Invalid plugin clientId");
+      return;
+    }
+
     this.clearPendingConnection(ws);
     pending.identity.clientId = clientId;
     if (message.appVersion) {
       pending.identity.appVersion = message.appVersion;
     }
-    const existing = this.externalSessionsByKey.get(clientId);
+    const existing = pluginId ? undefined : this.externalSessionsByKey.get(clientId);
     if (existing) {
-      this.incrementRuntimeCounter("helloResumed");
-      if (existing.externalDisconnectCleanupTimeout) {
-        clearTimeout(existing.externalDisconnectCleanupTimeout);
-        existing.externalDisconnectCleanupTimeout = null;
-      }
-      const newAppVersion = message.appVersion ?? null;
-      if (newAppVersion && newAppVersion !== existing.appVersion) {
-        existing.appVersion = newAppVersion;
-        existing.session.updateAppVersion(newAppVersion);
-      }
-      const newClientCapabilities = message.capabilities ?? null;
-      // COMPAT(selectiveAgentTimeline): added in v0.1.106. Every capable resumed
-      // hello resets membership before server_info so stale retained-session
-      // state cannot leak. Remove after 2027-01-12.
-      existing.session.updateClientCapabilities(newClientCapabilities, ws);
-      if (
-        JSON.stringify(existing.clientCapabilities ?? null) !==
-        JSON.stringify(newClientCapabilities ?? null)
-      ) {
-        existing.clientCapabilities = newClientCapabilities;
-        this.syncBrowserToolsClientRegistration(existing);
-      }
-      existing.sockets.add(ws);
-      this.sessions.set(ws, existing);
-      pending.identity.sessionId = existing.session.getSessionId();
-      this.syncBrowserToolsClientRegistration(existing);
-      this.sendToClient(ws, this.createServerInfoMessage());
-      pending.connectionLogger.info(
-        {
-          ...toConnectionLogFields(pending.identity),
-          resumed: true,
-          totalSessions: this.sessions.size,
-        },
-        "Client connected via hello",
-      );
+      this.resumeSession({ ws, message, pending, existing });
       return;
     }
 
@@ -1530,9 +1564,12 @@ export class VoiceAssistantWebSocketServer {
       appVersion: message.appVersion ?? null,
       clientCapabilities: message.capabilities ?? null,
       connectionLogger,
+      lifecycle: pluginId ? { kind: "ephemeral-plugin", pluginId } : { kind: "reconnectable" },
     });
     this.sessions.set(ws, connection);
-    this.externalSessionsByKey.set(clientId, connection);
+    if (connection.lifecycle === "reconnectable") {
+      this.externalSessionsByKey.set(clientId, connection);
+    }
     pending.identity.sessionId = connection.session.getSessionId();
     this.syncBrowserToolsClientRegistration(connection);
     this.sendToClient(ws, this.createServerInfoMessage());
@@ -1546,6 +1583,50 @@ export class VoiceAssistantWebSocketServer {
     );
   }
 
+  private resumeSession(params: {
+    ws: WebSocketLike;
+    message: WSHelloMessage;
+    pending: PendingConnection;
+    existing: ReconnectableSessionConnection;
+  }): void {
+    const { ws, message, pending, existing } = params;
+    this.incrementRuntimeCounter("helloResumed");
+    if (existing.externalDisconnectCleanupTimeout) {
+      clearTimeout(existing.externalDisconnectCleanupTimeout);
+      existing.externalDisconnectCleanupTimeout = null;
+    }
+    const newAppVersion = message.appVersion ?? null;
+    if (newAppVersion && newAppVersion !== existing.appVersion) {
+      existing.appVersion = newAppVersion;
+      existing.session.updateAppVersion(newAppVersion);
+    }
+    const newClientCapabilities = message.capabilities ?? null;
+    // COMPAT(selectiveAgentTimeline): added in v0.1.106. Every capable resumed
+    // hello resets membership before server_info so stale retained-session
+    // state cannot leak. Remove after 2027-01-12.
+    existing.session.updateClientCapabilities(newClientCapabilities, ws);
+    if (
+      JSON.stringify(existing.clientCapabilities ?? null) !==
+      JSON.stringify(newClientCapabilities ?? null)
+    ) {
+      existing.clientCapabilities = newClientCapabilities;
+      this.syncBrowserToolsClientRegistration(existing);
+    }
+    existing.sockets.add(ws);
+    this.sessions.set(ws, existing);
+    pending.identity.sessionId = existing.session.getSessionId();
+    this.syncBrowserToolsClientRegistration(existing);
+    this.sendToClient(ws, this.createServerInfoMessage());
+    pending.connectionLogger.info(
+      {
+        ...toConnectionLogFields(pending.identity),
+        resumed: true,
+        totalSessions: this.sessions.size,
+      },
+      "Client connected via hello",
+    );
+  }
+
   private buildServerInfoStatusPayload(): ServerInfoStatusPayload {
     return {
       status: "server_info",
@@ -1583,6 +1664,7 @@ export class VoiceAssistantWebSocketServer {
         // COMPAT(plugins): added in v0.3.0, remove gate after 2027-08-07.
         plugins: true,
         pluginManagement: true,
+        pluginLogs: true,
         // COMPAT(terminalRestoreModes): added in v0.1.81, remove gate after 2026-11-23.
         "terminal-restore-modes": true,
         // COMPAT(terminalInputModeReplay): added in v0.2.6, remove gate after 2027-02-02.
@@ -1756,6 +1838,7 @@ export class VoiceAssistantWebSocketServer {
         "Pending client disconnected",
       );
       this.socketIdentities.delete(ws);
+      this.finishPluginSocketCleanup(ws);
       return;
     }
 
@@ -1772,6 +1855,7 @@ export class VoiceAssistantWebSocketServer {
         );
         this.socketIdentities.delete(ws);
       }
+      this.finishPluginSocketCleanup(ws);
       return;
     }
 
@@ -1783,6 +1867,7 @@ export class VoiceAssistantWebSocketServer {
         "Hub session disconnected",
       );
       connection.session.cleanup();
+      this.finishPluginSocketCleanup(ws);
       return;
     }
     connection.sockets.delete(ws);
@@ -1791,6 +1876,12 @@ export class VoiceAssistantWebSocketServer {
 
     if (connection.sockets.size === 0) {
       this.unregisterBrowserToolsClient(connection.clientId);
+      if (connection.lifecycle === "ephemeral-plugin") {
+        this.pluginSocketIds.delete(ws);
+        await this.cleanupConnection(connection, "Plugin session disconnected");
+        this.finishPluginSocketCleanup(ws);
+        return;
+      }
       this.incrementRuntimeCounter("sessionDisconnectedWaitingReconnect");
       if (connection.externalDisconnectCleanupTimeout) {
         clearTimeout(connection.externalDisconnectCleanupTimeout);
@@ -1831,6 +1922,14 @@ export class VoiceAssistantWebSocketServer {
     }
 
     await this.cleanupConnection(connection, "Client disconnected");
+    this.finishPluginSocketCleanup(ws);
+  }
+
+  private finishPluginSocketCleanup(ws: WebSocketLike): void {
+    const resolve = this.pluginSocketCleanup.get(ws);
+    if (!resolve) return;
+    this.pluginSocketCleanup.delete(ws);
+    resolve();
   }
 
   private async cleanupConnection(
@@ -1838,7 +1937,7 @@ export class VoiceAssistantWebSocketServer {
     logMessage: string,
   ): Promise<void> {
     this.incrementRuntimeCounter("sessionCleanup");
-    if (connection.externalDisconnectCleanupTimeout) {
+    if (connection.lifecycle === "reconnectable" && connection.externalDisconnectCleanupTimeout) {
       clearTimeout(connection.externalDisconnectCleanupTimeout);
       connection.externalDisconnectCleanupTimeout = null;
     }
@@ -1848,9 +1947,11 @@ export class VoiceAssistantWebSocketServer {
       this.socketIdentities.delete(socket);
     }
     connection.sockets.clear();
-    const existing = this.externalSessionsByKey.get(connection.clientId);
-    if (existing === connection) {
-      this.externalSessionsByKey.delete(connection.clientId);
+    if (connection.lifecycle === "reconnectable") {
+      const existing = this.externalSessionsByKey.get(connection.clientId);
+      if (existing === connection) {
+        this.externalSessionsByKey.delete(connection.clientId);
+      }
     }
     this.unregisterBrowserToolsClient(connection.clientId);
 
@@ -2054,7 +2155,10 @@ export class VoiceAssistantWebSocketServer {
     ws: WebSocketLike,
     data: Buffer | ArrayBuffer | Buffer[] | string,
   ): void {
-    if (!this.acceptingConnections) {
+    if (
+      this.connectionLifecycle === "stopping" ||
+      (this.connectionLifecycle === "starting" && !this.pluginSocketIds.has(ws))
+    ) {
       return;
     }
 

packages/server/src/terminal/terminal-session-controller.test.ts

@@ -12,6 +12,7 @@ import type { ServerMessage, TerminalSession, TerminalStateSnapshot } from "./te
 import { TerminalSessionController } from "./terminal-session-controller.js";
 import type { TerminalManager, TerminalsChangedEvent } from "./terminal-manager.js";
 import { isSameOrDescendantPath } from "../server/path-utils.js";
+import { PluginSessionSocket } from "../server/plugins/session-socket.js";
 
 function deferred<T>(): {
   promise: Promise<T>;
@@ -755,6 +756,21 @@ describe("terminal-session-controller backpressure snapshot fallback", () => {
     expect(frames.some((frame) => frame.opcode === TerminalStreamOpcode.Output)).toBe(true);
   });
 
+  test("uses plugin IPC queued bytes to enter the snapshot backpressure path", async () => {
+    const socket = new PluginSessionSocket({
+      send() {
+        return true;
+      },
+    });
+    socket.send(new Uint8Array(8 * 1024 * 1024));
+    const { pushOutput, frames } = await setup(() => socket.bufferedAmount);
+
+    pushOutput("p".repeat(300 * 1024));
+    await waitForCoalescerFlush();
+
+    expect(frames.some((frame) => frame.opcode === TerminalStreamOpcode.Snapshot)).toBe(true);
+  });
+
   test("falls back to a snapshot at the byte threshold when no backpressure signal exists", async () => {
     // A null reading means the transport (e.g. the multiplexed relay socket) gives
     // no signal; we can't distinguish a slow client from a fast one, so we keep the