diff --git a/mcpb/server.js b/mcpb/server.js index dca9ca1..18baf5f 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,58 @@ 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; + + // Complete the handshake. If the server gates tool calls on + // `notifications/initialized` this is load-bearing; if not, it's harmless. + // Either way it's best-effort — a lost notification just means the replay + // 404s again and recovery terminates cleanly via the single-attempt cap in + // dispatch. Log a non-2xx (or a throw) so a required-but-failing + // notification is diagnosable from stderr instead of silently swallowed. + try { + const note = await postOnce({ jsonrpc: '2.0', method: 'notifications/initialized' }); + if (note.status !== 200 && note.status !== 202) { + process.stderr.write(`[obsidian-mcp-bridge] reinit: notifications/initialized → HTTP ${note.status}\n`); + } + await note.body?.cancel().catch(() => {}); + } catch (e) { + process.stderr.write(`[obsidian-mcp-bridge] reinit: notifications/initialized failed: ${e.message}\n`); + } + + // Fire-and-forget; startNotifyStream self-handles its own errors internally. + 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 +162,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; @@ -122,11 +171,35 @@ async function dispatch(message) { } } - if (response.status === 404 && sessionId) { + if (response.status === 404) { + // A 404 on /mcp means the server terminated/lost our session (idle GC or a + // server restart); per MCP spec the bridge should re-initialize, but some + // clients (e.g. Claude Desktop) don't act on the signal and the connection + // dead-ends. We deliberately do NOT gate on the bridge's current + // `sessionId`: a concurrent sibling's reinit may have already nulled it, + // and the 404 itself proves a session id was presented on this request + // (the server answers 400, handled below, when none is). So any 404 on a + // non-initialize call is a self-heal trigger. + // + // Self-heal once: re-establish a session and replay this exact message so + // recovery is invisible to the client (#238). Concurrent 404s coalesce + // onto a single reinit via the `initializing` gate and each replays under + // the fresh session. `isReplay` caps it at one attempt, so a session that + // dies again instantly terminates with -32000 rather than looping. + 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 +229,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..255ea39 --- /dev/null +++ b/tests/bridge-self-heal.test.ts @@ -0,0 +1,190 @@ +/** + * 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; initializing: 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('concurrent 404s coalesce onto a single reinit; each call replays under the fresh session', 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: {} } }); + } + if (rec.rpc === 'notifications/initialized') return res(202); + 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); + // Two tool calls in flight together: both 404 on sess-1 and must coalesce + // onto one re-init rather than each kicking off its own handshake. + const A = { jsonrpc: '2.0', id: 1, method: 'tools/call', params: { name: 'vault', arguments: { action: 'list' } } }; + const B = { jsonrpc: '2.0', id: 2, method: 'tools/call', params: { name: 'vault', arguments: { action: 'read' } } }; + await Promise.all([bridge.dispatch(A), bridge.dispatch(B)]); + + // Both clients get genuine results, neither sees the expiry error. + expect((emitted.find(m => m.id === 1) as { result?: { ok?: boolean } } | undefined)?.result?.ok).toBe(true); + expect((emitted.find(m => m.id === 2) as { result?: { ok?: boolean } } | undefined)?.result?.ok).toBe(true); + expect(emitted.some(m => (m as { error?: { code?: number } }).error?.code === -32000)).toBe(false); + // Exactly one self-heal despite two concurrent 404s: 1 initial + 1 reinit. + expect(calls.filter(c => c.rpc === 'initialize')).toHaveLength(2); + // Each tool call was attempted on the dead session and replayed on the new. + expect(calls.filter(c => c.rpc === 'tools/call' && c.session === 'sess-1')).toHaveLength(2); + expect(calls.filter(c => c.rpc === 'tools/call' && c.session === 'sess-2')).toHaveLength(2); + }); + + 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'); + }); +});