fix(server): start configured plugins concurrently (#8427)

The daemon accepts connections only after its configured plugins have
started, and PluginService.start started them one at a time, so each
plugin's build and process launch added to daemon startup. Configured
plugins now start together. Their providers and usage sources are still
published in config order: a plugin that finishes early waits for the
plugins listed before it, so when two plugins claim the same provider or
usage-source ID, the one listed first keeps it, as before. A failing
plugin still fails on its own.

Fixes #7979

packages/server/src/server/agent/idle-runtime-cleanup/idle-runtime-cleanup.test.ts

@@ -112,81 +112,24 @@ test.each(["foreground", "finish subscription"])(
   },
 );
 
-test("a provider veto retains the runtime until a fresh idle hour passes", async () => {
-  const s = scenario();
-  s.runtime.probe = async () => false;
-  expect(await s.sweep(61)).toEqual({ collected: [], failures: [] });
-  await s.sweep(119);
-  s.runtime.probe = async () => true;
-  expect((await s.sweep(120)).collected).toEqual([]);
-  expect((await s.sweep(179)).collected).toEqual([]);
-  expect((await s.sweep(180)).collected).toHaveLength(1);
-});
-
-test("repeated provider errors retry without postponing an already idle runtime", async () => {
-  const s = scenario();
-  const error = new Error("Probe failed");
-  s.runtime.probe = async () => {
-    throw error;
-  };
-  expect(await s.sweep(61)).toEqual({ collected: [], failures: [{ agentId: "agent", error }] });
-  expect(await s.sweep(119)).toEqual({ collected: [], failures: [{ agentId: "agent", error }] });
-  expect(s.runtime.closed).toBe(false);
-  s.runtime.probe = async () => true;
-  expect((await s.sweep(120)).collected).toEqual([
-    { agentId: "agent", provider: "test", sessionId: "agent" },
-  ]);
-});
-
-test("a failed probe does not clear a genuine veto's idle clock", async () => {
-  const s = scenario();
-  s.runtime.probe = async () => false;
-  await s.sweep(61);
-  s.runtime.probe = async () => {
-    throw new Error("Probe failed");
-  };
-  expect((await s.sweep(62)).failures).toHaveLength(1);
-  s.runtime.probe = async () => true;
-  expect((await s.sweep(63)).collected).toEqual([]);
-  expect((await s.sweep(122)).collected).toEqual([]);
-  expect((await s.sweep(123)).collected).toHaveLength(1);
-});
-
-test("a failed probe does not bypass the original one-hour idle period", async () => {
-  const s = scenario();
-  s.runtime.probe = async () => {
-    throw new Error("Probe failed");
-  };
-  expect((await s.sweep(1)).failures).toHaveLength(1);
-  s.runtime.probe = async () => true;
-  expect((await s.sweep(59)).collected).toEqual([]);
-  expect((await s.sweep(60)).collected).toHaveLength(1);
-});
-
-test("a child's failed probe retains its family only until safety is established", async () => {
-  const s = scenario();
-  const child = new Runtime("child");
-  child.parentAgentId = s.runtime.id;
-  const error = new Error("Child probe failed");
-  child.probe = async () => {
-    throw error;
-  };
-  s.runtimes.push(child);
-  expect(await s.sweep(61)).toEqual({
-    collected: [],
-    failures: [
-      { agentId: "agent", error },
-      { agentId: "child", error },
-    ],
-  });
-  expect(s.runtime.closed).toBe(false);
-  expect(child.closed).toBe(false);
-  child.probe = async () => true;
-  expect((await s.sweep(62)).collected).toEqual([
-    { agentId: "agent", provider: "test", sessionId: "agent" },
-    { agentId: "child", provider: "test", sessionId: "child" },
-  ]);
-});
+test.each(["veto", "error"])(
+  "a supported provider %s retains the runtime until a fresh idle hour passes",
+  async (decision) => {
+    const s = scenario();
+    s.runtime.probe = async () => {
+      if (decision === "error") throw new Error("Probe failed");
+      return false;
+    };
+    const blocked = await s.sweep(61);
+    expect(blocked.collected).toEqual([]);
+    expect(blocked.failures).toHaveLength(decision === "error" ? 1 : 0);
+    await s.sweep(119);
+    s.runtime.probe = async () => true;
+    expect((await s.sweep(120)).collected).toEqual([]);
+    expect((await s.sweep(179)).collected).toEqual([]);
+    expect((await s.sweep(180)).collected).toHaveLength(1);
+  },
+);
 
 test("an idle managed descendant with provider-held work vetoes its parent", async () => {
   const s = scenario();
@@ -214,7 +157,7 @@ test("new caller protection during an async provider probe prevents stale collec
   expect((await s.sweep(122)).collected).toHaveLength(1);
 });
 
-test("a provider timeout retains ownership until a fresh probe succeeds without restarting idleness", async () => {
+test("a supported provider timeout retains ownership even if it later returns true", async () => {
   const s = scenario();
   const probe = deferred<boolean>();
   s.runtime.probe = () => probe.promise;
@@ -222,13 +165,8 @@ test("a provider timeout retains ownership until a fresh probe succeeds without
   expect(result.collected).toEqual([]);
   expect(result.failures[0]?.error).toEqual(new Error("Runtime safety probe timed out for agent"));
   probe.resolve(true);
-  await probe.promise;
+  expect((await s.sweep(62)).collected).toEqual([]);
   expect(s.runtime.closed).toBe(false);
-  s.runtime.probe = async () => true;
-  expect((await s.sweep(62)).collected).toEqual([
-    { agentId: "agent", provider: "test", sessionId: "agent" },
-  ]);
-  expect(s.runtime.closed).toBe(true);
 }, 15_000);
 
 test("the host opt-out disables explicit sweeps and scheduled ticks", async () => {

packages/server/src/server/agent/idle-runtime-cleanup/index.ts

@@ -124,7 +124,7 @@ export function createIdleRuntimeCleanup(options: Options) {
         }
       } catch (error) {
         if (!isCurrent()) break;
-        // An unknown safety result retains ownership, but is not evidence of activity.
+        protectedAt.set(runtime.identity, { at: timestamp, protected: true });
         result.failures.push({ agentId: runtime.agentId, error });
         options.logger.warn(
           { agentId: runtime.agentId, err: error },

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

@@ -20,6 +20,7 @@ import { DaemonConfigStore } from "../daemon-config-store.js";
 import { PluginService } from "./index.js";
 import { BuiltinPluginLoader } from "./builtin/index.js";
 import { ManagedPluginSources } from "./managed-source.js";
+import type { PluginProviderMetadata } from "./plugin-process-protocol.js";
 import {
   startNpmRegistry,
   npmPluginPackages,
@@ -179,6 +180,55 @@ function createPluginSelectivePausedRuntime(pausedPluginId: string) {
   return { runtime, started, releaseStart, starts };
 }
 
+function createGatedRuntime(providers: Record<string, readonly PluginProviderMetadata[]> = {}) {
+  const gates = new Map<string, PromiseWithResolvers<void>>();
+  const gate = (pluginId: string) => {
+    const existing = gates.get(pluginId);
+    if (existing) return existing;
+    const created = Promise.withResolvers<void>();
+    gates.set(pluginId, created);
+    return created;
+  };
+  const begun: string[] = [];
+  const running = new Set<string>();
+  const runtime: TestPluginRuntime = {
+    ...emptyUsageRuntime,
+    catalog: () => [...running].map((id) => ({ id, clientBundle: "bundle" })),
+    invoke: async () => undefined,
+    getLogs: () => [],
+    clearLogs: () => undefined,
+    getProviderRegistrations: (pluginId) =>
+      running.has(pluginId) ? (providers[pluginId] ?? []) : [],
+    startPlugin: async (pluginId) => {
+      begun.push(pluginId);
+      await gate(pluginId).promise;
+      running.add(pluginId);
+    },
+    stopPluginById: async (pluginId) => running.delete(pluginId),
+    stopAll: async () => {
+      running.clear();
+    },
+    subscribe: () => () => undefined,
+    bindPaseoSessionHost: () => undefined,
+  };
+  return {
+    runtime,
+    begun,
+    finish: (pluginId: string) => gate(pluginId).resolve(),
+    fail: (pluginId: string, error: Error) => gate(pluginId).reject(error),
+  };
+}
+
+function nextNotification(service: PluginService, pluginId: string): Promise<void> {
+  return new Promise((resolve) => {
+    const unsubscribe = service.subscribe((notified) => {
+      if (notified !== pluginId) return;
+      unsubscribe();
+      resolve();
+    });
+  });
+}
+
 describe("PluginService", () => {
   it.each([
     { shadow: false, runtime: "subprocess" },
@@ -372,15 +422,127 @@ describe("PluginService", () => {
       "startup-first": { source: "directory", path: first },
       "startup-second": { source: "directory", path: second },
     });
-    const snapshots: string[][] = [];
-    service.subscribe(() => snapshots.push(catalogIds(service)));
+    const notifications: { pluginId: string; running: boolean }[] = [];
+    service.subscribe((pluginId) =>
+      notifications.push({ pluginId, running: catalogIds(service).includes(pluginId) }),
+    );
 
     await service.start();
 
-    expect(snapshots).toEqual([["startup-first"], ["startup-first", "startup-second"]]);
+    expect(notifications).toEqual([
+      { pluginId: "startup-first", running: true },
+      { pluginId: "startup-second", running: true },
+    ]);
+    expect(catalogIds(service)).toEqual(["startup-first", "startup-second"]);
     await service.stopAllPlugins();
   }, 20_000);
 
+  it("starts configured plugins concurrently and publishes them in config order", async () => {
+    const home = await mkdtemp(path.join(tmpdir(), "paseo-plugin-home-"));
+    roots.push(home);
+    const gated = createGatedRuntime();
+    const service = createService(
+      home,
+      {
+        alpha: { source: "directory", path: "/plugins/alpha" },
+        beta: { source: "directory", path: "/plugins/beta" },
+        gamma: { source: "directory", path: "/plugins/gamma" },
+      },
+      { runtime: gated.runtime },
+    );
+    const notified: string[] = [];
+    service.subscribe((pluginId) => notified.push(pluginId));
+    let started = false;
+    const starting = (async () => {
+      await service.start();
+      started = true;
+    })();
+
+    await expect.poll(() => gated.begun).toEqual(["alpha", "beta", "gamma"]);
+
+    gated.finish("gamma");
+    const alphaPublished = nextNotification(service, "alpha");
+    gated.finish("alpha");
+    await alphaPublished;
+    expect(notified).toEqual(["alpha"]);
+    expect(started).toBe(false);
+
+    gated.finish("beta");
+    await starting;
+    expect(notified).toEqual(["alpha", "beta", "gamma"]);
+    expect(catalogIds(service)).toEqual(["alpha", "beta", "gamma"]);
+    await service.stopAllPlugins();
+  });
+
+  it("starts the other configured plugins when one fails while they are still starting", async () => {
+    const home = await mkdtemp(path.join(tmpdir(), "paseo-plugin-home-"));
+    roots.push(home);
+    const gated = createGatedRuntime();
+    const service = createService(
+      home,
+      {
+        broken: { source: "directory", path: "/plugins/broken" },
+        healthy: { source: "directory", path: "/plugins/healthy" },
+      },
+      { runtime: gated.runtime },
+    );
+    const notified: string[] = [];
+    service.subscribe((pluginId) => notified.push(pluginId));
+    const starting = service.start();
+
+    await expect.poll(() => gated.begun).toEqual(["broken", "healthy"]);
+    const brokenPublished = nextNotification(service, "broken");
+    gated.fail("broken", new Error("startup exploded"));
+    await brokenPublished;
+    gated.finish("healthy");
+    await starting;
+
+    expect(notified).toEqual(["broken", "healthy"]);
+    expect(catalogIds(service)).toEqual(["healthy"]);
+    expect(await service.listPlugins()).toMatchObject([
+      { id: "broken", status: "failed", error: "startup exploded" },
+      { id: "healthy", status: "running" },
+    ]);
+    await service.stopAllPlugins();
+  });
+
+  it("keeps a provider ID with the plugin listed first when both start at once", async () => {
+    const home = await mkdtemp(path.join(tmpdir(), "paseo-plugin-home-"));
+    roots.push(home);
+    const gated = createGatedRuntime({
+      first: [{ id: "shared-agent", label: "First agent" }],
+      second: [{ id: "shared-agent", label: "Second agent" }],
+    });
+    const service = createService(
+      home,
+      {
+        first: { source: "directory", path: "/plugins/first" },
+        second: { source: "directory", path: "/plugins/second" },
+      },
+      { runtime: gated.runtime },
+    );
+    const starting = service.start();
+
+    await expect.poll(() => gated.begun).toEqual(["first", "second"]);
+    gated.finish("second");
+    gated.finish("first");
+    await starting;
+
+    expect(service.getProviderRegistrations()).toMatchObject([
+      { id: "shared-agent", label: "First agent" },
+    ]);
+    expect(catalogIds(service)).toEqual(["first"]);
+    expect(await service.listPlugins()).toMatchObject([
+      { id: "first", status: "running" },
+      {
+        id: "second",
+        status: "failed",
+        error: 'Plugin second cannot register provider ID "shared-agent" twice',
+      },
+    ]);
+    await service.stopAllPlugins();
+  });
+
   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);

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

@@ -230,11 +230,18 @@ export class PluginService {
       }
     });
     if (config.pluginsEnabled === true) {
-      for (const [pluginId, source] of Object.entries(config.plugins ?? {})) {
-        if (source.enabled === false) continue;
-        await this.startConfigured(pluginId);
-        this.notify(pluginId);
-      }
+      let earlierPublished: Promise<unknown> = Promise.resolve();
+      await Promise.all(
+        Object.entries(config.plugins ?? {})
+          .filter(([, source]) => source.enabled !== false)
+          .map(async ([pluginId]) => {
+            const publishTurn = earlierPublished;
+            const starting = this.startConfigured(pluginId, publishTurn);
+            earlierPublished = Promise.all([publishTurn, starting]);
+            await starting;
+            this.notify(pluginId);
+          }),
+      );
     }
     this.configStore.onFieldChange("pluginsEnabled", (value) => {
       this.handleGlobalSwitch(value === true);
@@ -560,12 +567,12 @@ export class PluginService {
     }).catch((error) => this.logger.error({ err: error }, "Failed to enable plugins"));
   }
 
-  private async startConfigured(pluginId: string): Promise<void> {
+  private async startConfigured(pluginId: string, publishTurn?: Promise<unknown>): Promise<void> {
     const source = this.configStore.get().plugins?.[pluginId];
     if (!source || source.enabled === false || !this.canPublish(pluginId)) return;
     this.errors.delete(pluginId);
     try {
-      await this.startPlugin(pluginId, source.path);
+      await this.startPlugin(pluginId, source.path, publishTurn);
     } catch (error) {
       if (this.canPublish(pluginId)) this.recordFailure(pluginId, error);
     }
@@ -594,8 +601,13 @@ export class PluginService {
     );
   }
 
-  private async startPlugin(pluginId: string, sourcePath: string): Promise<void> {
+  private async startPlugin(
+    pluginId: string,
+    sourcePath: string,
+    publishTurn?: Promise<unknown>,
+  ): Promise<void> {
     await this.runtime.startPlugin(pluginId, sourcePath, () => this.canPublish(pluginId));
+    await publishTurn;
     try {
       await this.publishProviderRegistrations(pluginId, sourcePath);
       this.publishUsageSources(pluginId);