feat: implement auto-start for STDIO connections and enhance event bus logging

- Added functionality to automatically start STDIO connections based on the AUTO_START_CONNECTIONS environment variable.
- Introduced a new helper function, autoStartConnectionsByTitle, to manage the auto-start process.
- Enhanced EventBus with additional logging for event processing and subscription matching.
- Implemented a mechanism to kill orphaned STDIO processes to prevent stale connections.
- Updated error handling in the proxy to avoid logging unnecessary "Method not found" errors.

This commit improves the management of STDIO connections and provides better visibility into event bus operations.

Author: vibegui

apps/mesh/src/api/app.ts

@@ -100,13 +100,72 @@ import {
 import { MiddlewareHandler } from "hono/types";
 import { getToolsByCategory, MANAGEMENT_TOOLS } from "../tools/registry";
 import { Env } from "./env";
+import { dangerouslyCreateSuperUserMCPProxy } from "./routes/proxy";
 import { resetStdioConnectionPool } from "../stdio/stable-transport";
 import { devLogger } from "./utils/dev-logger";
 
 const getHandleOAuthProtectedResourceMetadata = () =>
   oAuthProtectedResourceMetadata(auth);
 const getHandleOAuthDiscoveryMetadata = () => oAuthDiscoveryMetadata(auth);
 
+/**
+ * Auto-start STDIO connections by title
+ * Used with AUTO_START_CONNECTIONS env var
+ *
+ * Uses dangerouslyCreateSuperUserMCPProxy to create a system-level proxy
+ * that spawns the STDIO process with proper credentials.
+ */
+async function autoStartConnectionsByTitle(
+  database: MeshDatabase,
+  titles: string[],
+) {
+  const db = database.db;
+
+  // Query all STDIO connections matching the titles
+  const connections = await db
+    .selectFrom("connections")
+    .selectAll()
+    .where("connection_type", "=", "STDIO")
+    .where("title", "in", titles)
+    .execute();
+
+  if (connections.length === 0) {
+    console.log(`[AutoStart] No matching STDIO connections found`);
+    return;
+  }
+
+  console.log(
+    `[AutoStart] Found ${connections.length} connections to start: ${connections.map((c) => c.title).join(", ")}`,
+  );
+
+  for (const conn of connections) {
+    try {
+      console.log(`[AutoStart] Starting: ${conn.title} (${conn.id})`);
+
+      // Create system context and use the superuser proxy
+      const ctx = await ContextFactory.create();
+
+      const proxy = await dangerouslyCreateSuperUserMCPProxy(conn.id, {
+        ...ctx,
+        auth: { ...ctx.auth, user: { id: "auto-start" } },
+      });
+
+      // listTools() uses cached DB data, doesn't spawn STDIO
+      // listPrompts() forces actual client connection, triggering spawn
+      // Ignore "Method not found" - some MCPs don't implement prompts
+      try {
+        await proxy.client.listPrompts();
+      } catch {
+        // Ignore - the spawn happened, that's what matters
+      }
+
+      console.log(`[AutoStart] ✓ ${conn.title} started`);
+    } catch (error) {
+      console.error(`[AutoStart] ✗ ${conn.title} failed:`, error);
+    }
+  }
+}
+
 /**
  * Resource server metadata type
  */
@@ -136,7 +195,8 @@ export function createApp(options: CreateAppOptions = {}) {
 
   // Kill and respawn STDIO connections on restart/HMR
   // Old processes have stale credentials, need fresh spawn with new tokens
-  resetStdioConnectionPool().catch((err) => {
+  // IMPORTANT: Track this promise so autoStart waits for it to complete
+  const poolResetPromise = resetStdioConnectionPool().catch((err) => {
     console.error("[StableStdio] Error resetting pool:", err);
   });
 
@@ -494,6 +554,35 @@ export function createApp(options: CreateAppOptions = {}) {
     console.log("[EventBus] Worker started");
   });
 
+  // Auto-start connections specified in AUTO_START_CONNECTIONS env var
+  // Format: comma-separated connection titles, e.g. "Bridge,Pilot"
+  const autoStartConnections = process.env.AUTO_START_CONNECTIONS;
+  if (autoStartConnections) {
+    const connectionTitles = autoStartConnections
+      .split(",")
+      .map((s) => s.trim())
+      .filter(Boolean);
+    if (connectionTitles.length > 0) {
+      console.log(
+        `[AutoStart] Will start connections: ${connectionTitles.join(", ")}`,
+      );
+      // Wait for pool reset to complete before starting new connections
+      // This prevents race conditions where old processes haven't been killed yet
+      // Also add a small delay to let the app fully initialize
+      (async () => {
+        try {
+          // Wait for pool reset to finish (kills old STDIO processes)
+          await poolResetPromise;
+          // Small delay to ensure ports are released
+          await new Promise((resolve) => setTimeout(resolve, 500));
+          await autoStartConnectionsByTitle(database, connectionTitles);
+        } catch (error) {
+          console.error("[AutoStart] Failed:", error);
+        }
+      })();
+    }
+  }
+
   // Inject MeshContext into requests
   // Skip auth routes, static files, health check, and metrics - they don't need MeshContext
   app.use("*", async (c, next) => {

apps/mesh/src/api/routes/proxy.ts

@@ -376,6 +376,10 @@ async function createMCPProxyDoNotUseDirectly(
       env.MESH_URL = meshUrl;
     }
 
+    // Pass the connection ID so STDIO servers can identify themselves
+    // (needed for event bus subscriptions via gateway)
+    env.MESH_CONNECTION_ID = connectionId;
+
     // Pass state as JSON for bindings
     const state = connection.configuration_state;
     if (state && Object.keys(state).length > 0) {
@@ -648,7 +652,16 @@ async function createMCPProxyDoNotUseDirectly(
       client = await createClient();
       return await client.listPrompts();
     } catch (error) {
-      console.error("[proxy:listPrompts] Error listing prompts:", error);
+      // Prompts are optional in MCP - don't log "Method not found" errors
+      // Error code -32601 is "Method not found" which is expected for MCPs without prompts
+      const isMethodNotFound =
+        error &&
+        typeof error === "object" &&
+        "code" in error &&
+        error.code === -32601;
+      if (!isMethodNotFound) {
+        console.error("[proxy:listPrompts] Error listing prompts:", error);
+      }
       throw error;
     } finally {
       client?.close().catch(console.error);

apps/mesh/src/event-bus/event-bus.ts

@@ -118,6 +118,17 @@ export class EventBus implements IEventBus {
 
     // Find matching subscriptions and create delivery records
     const subscriptions = await this.storage.getMatchingSubscriptions(event);
+
+    console.log(
+      `[EventBus] Event ${event.type} from ${event.source}: ${subscriptions.length} matching subscriptions`,
+    );
+    if (subscriptions.length > 0) {
+      console.log(
+        `[EventBus] Subscribers:`,
+        subscriptions.map((s) => s.connectionId).join(", "),
+      );
+    }
+
     if (subscriptions.length > 0) {
       // Determine when to deliver:
       // - deliverAt: use specified time
@@ -134,6 +145,9 @@ export class EventBus implements IEventBus {
       // Only notify strategy for immediate delivery (no scheduled time and no cron)
       // Scheduled events will be picked up by the polling worker at the right time
       if (this.notifyStrategy && !deliverAt) {
+        console.log(
+          `[EventBus] Triggering immediate processing for ${event.type}`,
+        );
         await this.notifyStrategy.notify(eventId).catch((error) => {
           console.warn("[EventBus] Notify failed (non-critical):", error);
         });

apps/mesh/src/event-bus/worker.ts

@@ -109,6 +109,7 @@ export class EventBusWorker {
   private notifySubscriber: NotifySubscriberFn;
   private running = false;
   private processing = false;
+  private pendingProcessRequest = false;
   private config: Required<EventBusConfig>;
 
   constructor(
@@ -160,10 +161,15 @@ export class EventBusWorker {
    * Called by the NotifyStrategy when events are available
    */
   async processNow(): Promise<void> {
-    if (!this.running) return;
+    if (!this.running) {
+      return;
+    }
 
-    // Prevent concurrent processing
-    if (this.processing) return;
+    // If already processing, mark that we need another run after current finishes
+    if (this.processing) {
+      this.pendingProcessRequest = true;
+      return;
+    }
 
     this.processing = true;
     try {
@@ -172,6 +178,13 @@ export class EventBusWorker {
       console.error("[EventBus] Error processing events:", error);
     } finally {
       this.processing = false;
+
+      // If processNow was called while we were processing, run again
+      if (this.pendingProcessRequest) {
+        this.pendingProcessRequest = false;
+        // Use setImmediate to avoid stack overflow on rapid fire events
+        setImmediate(() => this.processNow());
+      }
     }
   }
 
@@ -186,6 +199,8 @@ export class EventBusWorker {
     );
     if (pendingDeliveries.length === 0) return;
 
+    console.log(`[EventBus] Processing ${pendingDeliveries.length} deliveries`);
+
     // Group by subscription (connection)
     const grouped = groupByConnection(pendingDeliveries);
 
@@ -194,12 +209,21 @@ export class EventBusWorker {
 
     for (const [subscriptionId, batch] of grouped) {
       try {
+        console.log(
+          `[EventBus] Delivering ${batch.events.length} events to ${batch.connectionId}`,
+        );
+
         // Call ON_EVENTS on the subscriber connection
         const result = await this.notifySubscriber(
           batch.connectionId,
           batch.events,
         );
 
+        console.log(
+          `[EventBus] Delivery result for ${batch.connectionId}:`,
+          JSON.stringify(result).slice(0, 200),
+        );
+
         // Check if per-event results were provided
         if (result.results && Object.keys(result.results).length > 0) {
           // Per-event mode: process each event individually

apps/mesh/src/stdio/stable-transport.ts

@@ -343,6 +343,51 @@ async function forceCloseAllStdioConnections(): Promise<void> {
   await new Promise((resolve) => setTimeout(resolve, 100));
 }
 
+/**
+ * Kill orphaned STDIO processes that might be left from previous Mesh instances.
+ * This handles cases where the connection pool is empty but old processes are still running.
+ */
+async function killOrphanedStdioProcesses(): Promise<void> {
+  const { exec } = await import("child_process");
+  const { promisify } = await import("util");
+  const execAsync = promisify(exec);
+
+  // Patterns for processes spawned by Mesh that might be orphaned
+  // These are common command patterns for STDIO MCPs
+  const patterns = [
+    "mesh-bridge.*server", // Mesh bridge server
+    "pilot.*server/main", // Pilot server
+  ];
+
+  for (const pattern of patterns) {
+    try {
+      // Use pkill with -f to match full command line
+      // -9 for SIGKILL to ensure termination
+      await execAsync(`pkill -9 -f "${pattern}" 2>/dev/null || true`);
+    } catch {
+      // Ignore errors - process might not exist
+    }
+  }
+
+  // Also kill anything listening on port 9999 (Bridge WebSocket)
+  try {
+    const { stdout } = await execAsync(`lsof -t -i:9999 2>/dev/null || true`);
+    const pids = stdout.trim().split("\n").filter(Boolean);
+    for (const pid of pids) {
+      try {
+        process.kill(Number(pid), "SIGKILL");
+        console.log(
+          `[StableStdio] Killed orphaned process on port 9999: PID ${pid}`,
+        );
+      } catch {
+        // Process might already be dead
+      }
+    }
+  } catch {
+    // Ignore errors
+  }
+}
+
 /**
  * Force close all connections and clear the pool
  * Used on app startup/HMR to ensure fresh processes with new credentials
@@ -352,6 +397,7 @@ export async function resetStdioConnectionPool(): Promise<void> {
     `[StableStdio] Reset requested. Pool size: ${connectionPool.size}, keys: [${Array.from(connectionPool.keys()).join(", ")}]`,
   );
 
+  // First, close connections we know about in the pool
   if (connectionPool.size > 0) {
     console.log(
       `[StableStdio] Resetting ${connectionPool.size} connections (killing processes)`,
@@ -360,9 +406,11 @@ export async function resetStdioConnectionPool(): Promise<void> {
     console.log(
       `[StableStdio] Reset complete. Pool size: ${connectionPool.size}`,
     );
-  } else {
-    console.log(`[StableStdio] Pool was empty, nothing to reset`);
   }
+
+  // Then, kill any orphaned processes that might be left from previous runs
+  // (handles case where pool was empty but old processes are still running)
+  await killOrphanedStdioProcesses();
 }
 
 // Register shutdown handlers - clean up connections before exit

apps/mesh/src/tools/eventbus/subscribe.ts

@@ -22,11 +22,12 @@ export const EVENT_SUBSCRIBE = defineTool({
     const organization = requireOrganization(ctx);
     await ctx.access.check();
 
-    // Get the subscriber connection ID from the caller's token
-    const connectionId = ctx.connectionId;
+    // Get the subscriber connection ID
+    // Use explicit subscriberId if provided (for calls via gateway), otherwise use caller's connection
+    const connectionId = input.subscriberId || ctx.connectionId;
     if (!connectionId) {
       throw new Error(
-        "Connection ID required to subscribe. Use a connection-scoped token.",
+        "Connection ID required to subscribe. Use a connection-scoped token or provide subscriberId.",
       );
     }
     // Create the subscription

packages/bindings/src/well-known/event-bus.ts

@@ -121,6 +121,18 @@ export const EventSubscribeInputSchema = z.object({
     .max(1000)
     .optional()
     .describe("JSONPath filter expression on event data"),
+
+  /**
+   * Optional: Override the subscriber connection ID.
+   * When calling through a gateway, use this to specify which connection
+   * should receive the events (defaults to the caller's connection ID).
+   */
+  subscriberId: z
+    .string()
+    .optional()
+    .describe(
+      "Override subscriber connection ID (for subscriptions via gateway)",
+    ),
 });
 
 export type EventSubscribeInput = z.infer<typeof EventSubscribeInputSchema>;