diff --git a/mcpb/server.js b/mcpb/server.js index dca9ca1..04a3470 100644 --- a/mcpb/server.js +++ b/mcpb/server.js @@ -7,20 +7,18 @@ const MCP_URL = process.env.MCP_URL; const API_KEY = process.env.MCP_API_KEY || ''; const FETCH_TIMEOUT_MS = 30_000; -if (!MCP_URL) { - process.stderr.write('[obsidian-mcp-bridge] MCP_URL not set\n'); - process.exit(1); -} -try { new URL(MCP_URL); } catch { - process.stderr.write(`[obsidian-mcp-bridge] invalid MCP_URL: ${MCP_URL}\n`); - process.exit(1); -} - let sessionId = null; let notifyAbort = null; // Serializes dispatch until initialize lands a session id. Subsequent // messages chain off this promise so they don't race past the handshake. +// Also gates an in-flight self-heal re-initialize so concurrent 404s don't +// each kick off their own handshake. let initializing = null; +// The last `initialize` we forwarded, kept so the bridge can transparently +// re-establish a session if the server evicts/loses it (issue #238). Replayed +// with a synthetic id and its result suppressed — the client never sees it. +let cachedInitialize = null; +let reinitCounter = 0; function headers(extra = {}) { const h = { @@ -96,9 +94,48 @@ async function postOnce(message) { }); } -async function dispatch(message) { +// Transparently re-establish a session the server has evicted (#238). Replays +// the cached `initialize` (with a fresh synthetic id, result discarded), then +// completes the handshake with `notifications/initialized` so the new session +// is ready for tool calls. Returns true once a fresh session id is in hand. +// Never emits to the client — the caller replays the original message and emits +// that result, so recovery is invisible. +async function reinitialize() { + if (!cachedInitialize) return false; + // Drop the dead session so postOnce sends the initialize with no session id. + sessionId = null; + if (notifyAbort) { notifyAbort.abort(); notifyAbort = null; } + + let response; + try { + response = await postOnce({ ...cachedInitialize, id: `reinit-${reinitCounter++}` }); + } catch (e) { + process.stderr.write(`[obsidian-mcp-bridge] reinit fetch failed: ${e.message}\n`); + return false; + } + const sid = response.headers.get('mcp-session-id'); + // We only need the session id from the headers; discard the body unread. + await response.body?.cancel().catch(() => {}); + if (!response.ok || !sid) { + process.stderr.write(`[obsidian-mcp-bridge] reinit failed: HTTP ${response.status}${sid ? '' : ', no session id'}\n`); + return false; + } + sessionId = sid; + + // Best-effort: a stateful server expects `notifications/initialized` before + // it will service tool calls. It's a notification (202, no body to emit). + try { + const note = await postOnce({ jsonrpc: '2.0', method: 'notifications/initialized' }); + await note.body?.cancel().catch(() => {}); + } catch { /* server may not require it; the replay will reveal the truth */ } + + startNotifyStream(); + return true; +} + +async function dispatch(message, isReplay = false) { // If we haven't established a session yet and this isn't the initialize, - // wait for the in-flight initialize to land first. + // wait for the in-flight initialize (or self-heal re-init) to land first. if (!sessionId && initializing && message.method !== 'initialize') { try { await initializing; } catch { /* let dispatch surface its own error below */ } } @@ -115,6 +152,8 @@ async function dispatch(message) { } if (message.method === 'initialize') { + // Cache the handshake so we can transparently replay it on session loss. + cachedInitialize = message; const sid = response.headers.get('mcp-session-id'); if (sid) { sessionId = sid; @@ -123,10 +162,26 @@ async function dispatch(message) { } if (response.status === 404 && sessionId) { + // The server evicted/lost our session (idle GC, or a server restart). Per + // MCP spec the 404 means "re-initialize," but some clients (e.g. Claude + // Desktop) don't act on it and the connection dead-ends. Self-heal once: + // re-establish a session and replay this exact message so recovery is + // invisible to the client (#238). `isReplay` caps it at a single attempt + // so a session that dies again instantly can't spin a reinit loop. + if (!isReplay && message.method !== 'initialize' && cachedInitialize) { + if (!initializing) { + initializing = reinitialize().finally(() => { initializing = null; }); + } + let healed = false; + try { healed = await initializing; } catch { healed = false; } + if (healed) { + return await dispatch(message, true); + } + } sessionId = null; if (notifyAbort) { notifyAbort.abort(); notifyAbort = null; } if (message.id != null) { - emit({ jsonrpc: '2.0', id: message.id, error: { code: -32000, message: 'Session expired; please reinitialize' } }); + emit({ jsonrpc: '2.0', id: message.id, error: { code: -32000, message: 'Session expired and automatic reinitialize failed; please retry.' } }); } return; } @@ -156,42 +211,77 @@ async function dispatch(message) { } } -const rl = createInterface({ input: process.stdin }); -rl.on('line', (line) => { - const trimmed = line.trim(); - if (!trimmed) return; - let msg; - try { msg = JSON.parse(trimmed); } - catch (e) { - process.stderr.write(`[obsidian-mcp-bridge] stdin parse: ${e.message}\n`); - return; +// Stdio bootstrap. Guarded so the module can be `require`d in tests (which +// drive `dispatch` directly) without validating env, claiming stdin, or +// exiting the process. When launched as the bridge, `require.main === module`. +function main() { + if (!MCP_URL) { + process.stderr.write('[obsidian-mcp-bridge] MCP_URL not set\n'); + process.exit(1); } - // Capture the initialize promise so concurrent messages can wait for the - // session id before posting. - if (msg.method === 'initialize') { - initializing = dispatch(msg).finally(() => { initializing = null; }); - initializing.catch(err => { - process.stderr.write(`[obsidian-mcp-bridge] dispatch: ${err.message}\n`); - }); - } else { - dispatch(msg).catch(err => { - process.stderr.write(`[obsidian-mcp-bridge] dispatch: ${err.message}\n`); - }); + try { new URL(MCP_URL); } catch { + process.stderr.write(`[obsidian-mcp-bridge] invalid MCP_URL: ${MCP_URL}\n`); + process.exit(1); } -}); -process.stdin.on('end', async () => { - if (notifyAbort) notifyAbort.abort(); - // Best-effort: tell the server to retire the session so it doesn't linger - // in the pool until idle GC. - if (sessionId) { - try { - await fetch(MCP_URL, { - method: 'DELETE', - headers: headers(), - signal: AbortSignal.timeout(5_000), + const rl = createInterface({ input: process.stdin }); + rl.on('line', (line) => { + const trimmed = line.trim(); + if (!trimmed) return; + let msg; + try { msg = JSON.parse(trimmed); } + catch (e) { + process.stderr.write(`[obsidian-mcp-bridge] stdin parse: ${e.message}\n`); + return; + } + // Capture the initialize promise so concurrent messages can wait for the + // session id before posting. + if (msg.method === 'initialize') { + initializing = dispatch(msg).finally(() => { initializing = null; }); + initializing.catch(err => { + process.stderr.write(`[obsidian-mcp-bridge] dispatch: ${err.message}\n`); }); - } catch { /* server may have already gone; ignore */ } - } - process.exit(0); -}); + } else { + dispatch(msg).catch(err => { + process.stderr.write(`[obsidian-mcp-bridge] dispatch: ${err.message}\n`); + }); + } + }); + + process.stdin.on('end', async () => { + if (notifyAbort) notifyAbort.abort(); + // Best-effort: tell the server to retire the session so it doesn't linger + // in the pool until idle GC. + if (sessionId) { + try { + await fetch(MCP_URL, { + method: 'DELETE', + headers: headers(), + signal: AbortSignal.timeout(5_000), + }); + } catch { /* server may have already gone; ignore */ } + } + process.exit(0); + }); +} + +if (require.main === module) { + main(); +} + +// Test seam (inert when run as the bridge): expose dispatch and a state reset +// so the self-heal path can be exercised without spawning a process. +module.exports = { + dispatch, + reinitialize, + __reset() { + sessionId = null; + notifyAbort = null; + initializing = null; + cachedInitialize = null; + reinitCounter = 0; + }, + __state() { + return { sessionId, cachedInitialize, initializing }; + }, +}; diff --git a/tests/bridge-self-heal.test.ts b/tests/bridge-self-heal.test.ts new file mode 100644 index 0000000..cc194fc --- /dev/null +++ b/tests/bridge-self-heal.test.ts @@ -0,0 +1,160 @@ +/** + * Regression harness for #238: the .mcpb stdio bridge (mcpb/server.js) must + * self-heal when the server evicts/loses a session. + * + * Before the fix, a 404 on a non-initialize call made the bridge clear its + * session and emit `-32000 "please reinitialize"` to the client; clients that + * ignore the signal (e.g. Claude Desktop) then sent the next tool call with no + * session → HTTP 400 → the connection dead-ended until a manual restart. + * + * The bridge now caches the `initialize` handshake and, on a 404, transparently + * re-initializes and replays the original message once, so recovery is + * invisible to the client. These tests drive `dispatch` directly with a mocked + * `fetch`, capturing what the bridge writes to stdout. + */ + +// Must be set before the bridge module is required (it reads env at load). +process.env.MCP_URL = 'http://localhost:3001/mcp'; +process.env.MCP_API_KEY = 'test-key'; + +// eslint-disable-next-line @typescript-eslint/no-require-imports -- bridge is plain CommonJS, loaded as a module to test dispatch +const bridge = require('../mcpb/server.js') as { + dispatch: (msg: unknown, isReplay?: boolean) => Promise; + __reset: () => void; + __state: () => { sessionId: string | null; cachedInitialize: unknown }; +}; + +interface FakeResponse { + ok: boolean; + status: number; + headers: { get: (k: string) => string | null }; + text: () => Promise; + body: { cancel: () => Promise; getReader: () => { read: () => Promise<{ done: boolean; value: undefined }> } }; +} + +function res(status: number, opts: { sid?: string; json?: unknown } = {}): FakeResponse { + const h = new Map(); + if (opts.sid) h.set('mcp-session-id', opts.sid); + h.set('content-type', opts.json !== undefined ? 'application/json' : 'text/plain'); + return { + ok: status >= 200 && status < 300, + status, + headers: { get: (k: string) => h.get(k.toLowerCase()) ?? null }, + text: async () => (opts.json !== undefined ? JSON.stringify(opts.json) : ''), + body: { + cancel: async () => {}, + getReader: () => ({ read: async () => ({ done: true, value: undefined }) }), + }, + }; +} + +interface FetchRecord { method: string; session?: string; rpc?: string; id?: unknown } + +const INIT = { jsonrpc: '2.0', id: 0, method: 'initialize', params: { protocolVersion: '2025-06-18', capabilities: {}, clientInfo: { name: 't', version: '0' } } }; +const TOOL_CALL = { jsonrpc: '2.0', id: 1, method: 'tools/call', params: { name: 'vault', arguments: { action: 'list' } } }; + +describe('mcpb bridge self-heal on session expiry (#238)', () => { + let emitted: Array>; + let calls: FetchRecord[]; + let writeSpy: jest.SpyInstance; + + function installFetch(impl: (rec: FetchRecord, opts: { method: string; headers?: Record; body?: string }) => FakeResponse) { + (global as unknown as { fetch: unknown }).fetch = jest.fn(async (_url: string, opts: { method: string; headers?: Record; body?: string }) => { + const session = opts.headers?.['Mcp-Session-Id']; + if (opts.method === 'GET' || opts.method === 'DELETE') { + calls.push({ method: opts.method, session }); + return res(200); + } + const msg = JSON.parse(opts.body as string) as { method: string; id?: unknown }; + const rec: FetchRecord = { method: opts.method, session, rpc: msg.method, id: msg.id }; + calls.push(rec); + return impl(rec, opts); + }); + } + + beforeEach(() => { + bridge.__reset(); + emitted = []; + calls = []; + // emit() writes JSON + '\n' to stdout; capture and parse each line. + writeSpy = jest.spyOn(process.stdout, 'write').mockImplementation((chunk: string | Uint8Array) => { + const text = typeof chunk === 'string' ? chunk : Buffer.from(chunk).toString('utf8'); + for (const line of text.split('\n')) { + if (line.trim()) emitted.push(JSON.parse(line)); + } + return true; + }); + }); + + afterEach(() => { + writeSpy.mockRestore(); + }); + + test('404 on a tool call → bridge re-initializes and replays, emitting the real result (not an error)', async () => { + let sidSeq = 0; + installFetch((rec) => { + if (rec.rpc === 'initialize') { + sidSeq += 1; + return res(200, { sid: `sess-${sidSeq}`, json: { jsonrpc: '2.0', id: rec.id, result: { serverInfo: { name: 'x' } } } }); + } + if (rec.rpc === 'notifications/initialized') return res(202); + // First session (sess-1) is the evicted one; the replay uses sess-2. + if (rec.session === 'sess-1') return res(404, { sid: 'sess-1', json: { jsonrpc: '2.0', id: rec.id, error: { code: -32001, message: 'expired' } } }); + return res(200, { json: { jsonrpc: '2.0', id: rec.id, result: { ok: true } } }); + }); + + await bridge.dispatch(INIT); + await bridge.dispatch(TOOL_CALL); + + // The client receives the genuine tool result, never the expiry error. + const toolResult = emitted.find(m => m.id === 1); + expect(toolResult).toBeDefined(); + expect((toolResult as { result?: { ok?: boolean } }).result?.ok).toBe(true); + expect(emitted.some(m => (m as { error?: { code?: number } }).error?.code === -32000)).toBe(false); + + // It re-initialized once and replayed the tool call under the fresh session. + const reinit = calls.filter(c => c.rpc === 'initialize'); + expect(reinit).toHaveLength(2); + const toolCalls = calls.filter(c => c.rpc === 'tools/call'); + expect(toolCalls.map(c => c.session)).toEqual(['sess-1', 'sess-2']); + }); + + test('reinitialize itself fails → bridge surfaces a single -32000 and does not replay or loop', async () => { + let initCount = 0; + installFetch((rec) => { + if (rec.rpc === 'initialize') { + initCount += 1; + // First handshake establishes sess-1; the self-heal handshake fails + // (200 but no session id), so recovery must give up — not loop. + if (initCount === 1) return res(200, { sid: 'sess-1', json: { jsonrpc: '2.0', id: rec.id, result: {} } }); + return res(200, { json: { jsonrpc: '2.0', id: rec.id, result: {} } }); + } + if (rec.rpc === 'notifications/initialized') return res(202); + return res(404, { sid: rec.session, json: { jsonrpc: '2.0', id: rec.id, error: { code: -32001, message: 'expired' } } }); + }); + + await bridge.dispatch(INIT); + await bridge.dispatch(TOOL_CALL); + + const err = emitted.find(m => m.id === 1) as { error?: { code?: number; message?: string } } | undefined; + expect(err?.error?.code).toBe(-32000); + expect(err?.error?.message).toMatch(/reinitialize failed/i); + // Exactly one self-heal attempt: 2 initialize calls total, never a third. + expect(calls.filter(c => c.rpc === 'initialize')).toHaveLength(2); + // The original tool call was attempted once; the failed heal means no + // successful replay, and the cap prevents a reinit loop. + expect(calls.filter(c => c.rpc === 'tools/call')).toHaveLength(1); + }); + + test('initialize handshake is cached so a later session loss can be replayed', async () => { + installFetch((rec) => { + if (rec.rpc === 'initialize') return res(200, { sid: 'sess-1', json: { jsonrpc: '2.0', id: rec.id, result: {} } }); + return res(200, { json: { jsonrpc: '2.0', id: rec.id, result: {} } }); + }); + + expect(bridge.__state().cachedInitialize).toBeNull(); + await bridge.dispatch(INIT); + expect(bridge.__state().cachedInitialize).toMatchObject({ method: 'initialize' }); + expect(bridge.__state().sessionId).toBe('sess-1'); + }); +});