diff --git a/gitnexus/src/mcp/compatible-stdio-transport.ts b/gitnexus/src/mcp/compatible-stdio-transport.ts index b5fc856304..fa8d9f4d2d 100644 --- a/gitnexus/src/mcp/compatible-stdio-transport.ts +++ b/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) { diff --git a/gitnexus/src/mcp/server.ts b/gitnexus/src/mcp/server.ts index 4696bbba6c..b2bd5530f5 100644 --- a/gitnexus/src/mcp/server.ts +++ b/gitnexus/src/mcp/server.ts @@ -337,7 +337,6 @@ export async function startMCPServer(backend: LocalBackend): Promise { }, }); 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 { }); // 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); } diff --git a/gitnexus/test/unit/compatible-stdio-transport.test.ts b/gitnexus/test/unit/compatible-stdio-transport.test.ts index 4d2a148b47..d5686e3165 100644 --- a/gitnexus/test/unit/compatible-stdio-transport.test.ts +++ b/gitnexus/test/unit/compatible-stdio-transport.test.ts @@ -9,6 +9,20 @@ function onceMessage(transport: CompatibleStdioServerTransport): Promise { }); } +function trackClose(transport: CompatibleStdioServerTransport): { + onclose: ReturnType; + closed: Promise; +} { + const onclose = vi.fn(); + const closed = new Promise((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); + }); }); diff --git a/gitnexus/test/unit/server.test.ts b/gitnexus/test/unit/server.test.ts index d5ff9d7307..f2ac995da6 100644 --- a/gitnexus/test/unit/server.test.ts +++ b/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)', () => {