Merge pull request #243 from aaronsb/fix/bridge-self-heal-404

fix(bridge): self-heal on session expiry — reinitialize + replay (#238)

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 };
+  },
+};

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<void>;
+  __reset: () => void;
+  __state: () => { sessionId: string | null; cachedInitialize: unknown; initializing: unknown };
+};
+
+interface FakeResponse {
+  ok: boolean;
+  status: number;
+  headers: { get: (k: string) => string | null };
+  text: () => Promise<string>;
+  body: { cancel: () => Promise<void>; getReader: () => { read: () => Promise<{ done: boolean; value: undefined }> } };
+}
+
+function res(status: number, opts: { sid?: string; json?: unknown } = {}): FakeResponse {
+  const h = new Map<string, string>();
+  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<Record<string, unknown>>;
+  let calls: FetchRecord[];
+  let writeSpy: jest.SpyInstance;
+
+  function installFetch(impl: (rec: FetchRecord, opts: { method: string; headers?: Record<string, string>; body?: string }) => FakeResponse) {
+    (global as unknown as { fetch: unknown }).fetch = jest.fn(async (_url: string, opts: { method: string; headers?: Record<string, string>; 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');
+  });
+});