fix(mcp): prevent orphan processes by handling stdin close/end and startup race condition

gitnexus/src/mcp/compatible-stdio-transport.ts

@@ -48,12 +48,16 @@ const MAX_BUFFER_SIZE = 10 * 1024 * 1024; // 10 MB — generous for JSON-RPC
 export class CompatibleStdioServerTransport implements Transport {
   private _readBuffer: Buffer | undefined;
   private _started = false;
+  private _closed = false;
   private _framing: StdioFraming | null = null;
 
   onmessage?: (message: JSONRPCMessage) => void;
   onerror?: (error: Error) => void;
   onclose?: () => void;
 
+  /** Stable bound reference so `on`/`off` always use the same function. */
+  private readonly _boundClose = this.close.bind(this);
+
   constructor(
     private readonly _stdin: NodeJS.ReadableStream = process.stdin,
     private readonly _stdout: NodeJS.WritableStream = process.stdout,
@@ -77,10 +81,27 @@ export class CompatibleStdioServerTransport implements Transport {
     if (this._started) {
       throw new Error('CompatibleStdioServerTransport already started!');
     }
+    if (this._closed) {
+      throw new Error('CompatibleStdioServerTransport has been permanently closed');
+    }
+
+    // If stdin is already closed (parent died before we registered listeners),
+    // the 'end'/'close' event was already emitted and will never fire again.
+    // Close the transport immediately to avoid orphaned processes.
+    if (
+      (this._stdin as NodeJS.ReadStream).readableEnded ||
+      (this._stdin as NodeJS.ReadStream).destroyed
+    ) {
+      this._started = true;
+      await this.close();
+      return;
+    }
 
     this._started = true;
     this._stdin.on('data', this._ondata);
     this._stdin.on('error', this._onerror);
+    this._stdin.on('end', this._boundClose);
+    this._stdin.on('close', this._boundClose);
   }
 
   private detectFraming(): StdioFraming | null {
@@ -201,8 +222,13 @@ export class CompatibleStdioServerTransport implements Transport {
   }
 
   async close() {
+    if (this._closed) return;
+    this._closed = true;
+
     this._stdin.off('data', this._ondata);
     this._stdin.off('error', this._onerror);
+    this._stdin.off('end', this._boundClose);
+    this._stdin.off('close', this._boundClose);
 
     const remainingDataListeners = this._stdin.listenerCount('data');
     if (remainingDataListeners === 0) {

gitnexus/src/mcp/server.ts

@@ -337,7 +337,6 @@ export async function startMCPServer(backend: LocalBackend): Promise<void> {
     },
   });
   const transport = new CompatibleStdioServerTransport(process.stdin, safeStdout);
-  await server.connect(transport);
 
   // Surface the redirect counter on shutdown so users see the volume of
   // stray writes even when individual payloads were truncated/suppressed.
@@ -389,9 +388,20 @@ export async function startMCPServer(backend: LocalBackend): Promise<void> {
   });
 
   // Handle stdio errors — stdin close means the parent process is gone.
+  // Defense-in-depth: the transport also listens for stdin end/close and
+  // handles transport-level cleanup. These listeners handle process-level
+  // shutdown. Both paths are idempotent and safe to fire together.
   // Wrap so the event payload (e.g. an Error for 'error') can never reach
   // process.exit() as a non-numeric exit code, and void the returned promise.
   process.stdin.on('end', () => void shutdown(0));
+  process.stdin.on('close', () => void shutdown(0));
   process.stdin.on('error', () => void shutdown(0));
   process.stdout.on('error', () => void shutdown(0));
+
+  if (process.stdin.readableEnded || process.stdin.destroyed) {
+    await shutdown(0);
+    return;
+  }
+
+  await server.connect(transport);
 }

gitnexus/test/unit/compatible-stdio-transport.test.ts

@@ -9,6 +9,20 @@ function onceMessage(transport: CompatibleStdioServerTransport): Promise<any> {
   });
 }
 
+function trackClose(transport: CompatibleStdioServerTransport): {
+  onclose: ReturnType<typeof vi.fn>;
+  closed: Promise<void>;
+} {
+  const onclose = vi.fn();
+  const closed = new Promise<void>((resolve) => {
+    transport.onclose = () => {
+      onclose();
+      resolve();
+    };
+  });
+  return { onclose, closed };
+}
+
 describe('CompatibleStdioServerTransport', () => {
   let stdin: PassThrough;
   let stdout: PassThrough;
@@ -258,4 +272,57 @@ describe('CompatibleStdioServerTransport', () => {
 
     expect(raw).toBe('{"jsonrpc":"2.0","id":1,"result":{"ok":true}}\n');
   });
+
+  // ─── stdin lifecycle behavior ──────────────────────────────
+
+  it('stdin end closes transport and calls onclose exactly once', async () => {
+    const { onclose, closed } = trackClose(transport);
+
+    await transport.start();
+    stdin.push(null);
+
+    await closed;
+    expect(onclose).toHaveBeenCalledTimes(1);
+  });
+
+  it('stdin close closes transport and calls onclose exactly once', async () => {
+    const { onclose, closed } = trackClose(transport);
+
+    await transport.start();
+    stdin.destroy();
+
+    await closed;
+    expect(onclose).toHaveBeenCalledTimes(1);
+  });
+
+  it('close() is idempotent: calling close twice calls onclose once', async () => {
+    const onclose = vi.fn();
+    transport.onclose = onclose;
+
+    await transport.start();
+    await transport.close();
+    await transport.close();
+
+    expect(onclose).toHaveBeenCalledTimes(1);
+  });
+
+  it('start() immediately closes if stdin is already ended/destroyed before listeners register', async () => {
+    const endedStdin = new PassThrough();
+    endedStdin.push(null);
+
+    const t = new CompatibleStdioServerTransport(endedStdin, new PassThrough());
+    const { onclose, closed } = trackClose(t);
+
+    await t.start();
+
+    await closed;
+    expect(onclose).toHaveBeenCalledTimes(1);
+  });
+
+  it('start() after close throws', async () => {
+    await transport.start();
+    await transport.close();
+
+    await expect(transport.start()).rejects.toThrow(/close/i);
+  });
 });

gitnexus/test/unit/server.test.ts

@@ -14,10 +14,12 @@
  */
 import { describe, it, expect, vi } from 'vitest';
 import { Client } from '@modelcontextprotocol/sdk/client/index.js';
+import { Server } from '@modelcontextprotocol/sdk/server/index.js';
 import { InMemoryTransport } from '@modelcontextprotocol/sdk/inMemory.js';
 import {
   createMCPServer,
   installSignalShutdown,
+  startMCPServer,
   SHUTDOWN_EXIT_CODES,
 } from '../../src/mcp/server.js';
 import { GITNEXUS_TOOLS } from '../../src/mcp/tools.js';
@@ -130,6 +132,90 @@ describe('prompt registration', () => {
   });
 });
 
+// ─── startMCPServer ────────────────────────────────────────────────────
+
+describe('startMCPServer', () => {
+  it('registers stdin shutdown listeners before awaiting Server.prototype.connect', async () => {
+    // P1 finding: startMCPServer must register process.stdin end/close/error
+    // shutdown listeners before calling server.connect(), so that a stdin
+    // closure during transport setup is safely handled.
+    // This test asserts the listeners are present *during* the connect call.
+
+    // Capture baseline listener counts before startMCPServer adds anything
+    const closeBefore = process.stdin.listenerCount('close');
+    const endBefore = process.stdin.listenerCount('end');
+    const errorBefore = process.stdin.listenerCount('error');
+    const beforeListeners = {
+      stdinEnd: new Set(process.stdin.listeners('end')),
+      stdinClose: new Set(process.stdin.listeners('close')),
+      stdinError: new Set(process.stdin.listeners('error')),
+      stdoutError: new Set(process.stdout.listeners('error')),
+      sigint: new Set(process.listeners('SIGINT')),
+      sigterm: new Set(process.listeners('SIGTERM')),
+      exit: new Set(process.listeners('exit')),
+      uncaughtException: new Set(process.listeners('uncaughtException')),
+      unhandledRejection: new Set(process.listeners('unhandledRejection')),
+    };
+
+    // Stub process.exit to prevent actual termination if shutdown fires
+    const exitStub = vi.spyOn(process, 'exit').mockImplementation(() => undefined as never);
+
+    // Spy on Server.prototype.connect to inspect listener counts during the call
+    const connectSpy = vi.spyOn(Server.prototype, 'connect').mockImplementation(async function () {
+      expect(process.stdin.listenerCount('close')).toBeGreaterThan(closeBefore);
+      expect(process.stdin.listenerCount('end')).toBeGreaterThan(endBefore);
+      expect(process.stdin.listenerCount('error')).toBeGreaterThan(errorBefore);
+    });
+
+    const backend = createMockBackend();
+
+    try {
+      await startMCPServer(backend);
+    } finally {
+      connectSpy.mockRestore();
+      exitStub.mockRestore();
+      // Remove only listeners added by startMCPServer so no side effects leak.
+      for (const listener of process.stdin.listeners('end')) {
+        if (!beforeListeners.stdinEnd.has(listener)) process.stdin.removeListener('end', listener);
+      }
+      for (const listener of process.stdin.listeners('close')) {
+        if (!beforeListeners.stdinClose.has(listener)) {
+          process.stdin.removeListener('close', listener);
+        }
+      }
+      for (const listener of process.stdin.listeners('error')) {
+        if (!beforeListeners.stdinError.has(listener)) {
+          process.stdin.removeListener('error', listener);
+        }
+      }
+      for (const listener of process.stdout.listeners('error')) {
+        if (!beforeListeners.stdoutError.has(listener)) {
+          process.stdout.removeListener('error', listener);
+        }
+      }
+      for (const listener of process.listeners('SIGINT')) {
+        if (!beforeListeners.sigint.has(listener)) process.removeListener('SIGINT', listener);
+      }
+      for (const listener of process.listeners('SIGTERM')) {
+        if (!beforeListeners.sigterm.has(listener)) process.removeListener('SIGTERM', listener);
+      }
+      for (const listener of process.listeners('exit')) {
+        if (!beforeListeners.exit.has(listener)) process.removeListener('exit', listener);
+      }
+      for (const listener of process.listeners('uncaughtException')) {
+        if (!beforeListeners.uncaughtException.has(listener)) {
+          process.removeListener('uncaughtException', listener);
+        }
+      }
+      for (const listener of process.listeners('unhandledRejection')) {
+        if (!beforeListeners.unhandledRejection.has(listener)) {
+          process.removeListener('unhandledRejection', listener);
+        }
+      }
+    }
+  });
+});
+
 // ─── Graceful shutdown signal handling (#1132) ────────────────────────
 
 describe('installSignalShutdown (#1132)', () => {