pediatric-ai-scribe-v3/src/utils/clinicalMcpClient.js

var axios = require('axios');
var sleep = require('node:timers/promises').setTimeout;

var MCP_URLS = buildMcpUrls();
var _lastGoodMcpUrl = MCP_URLS[0];
// The server names this tool clinical_semantic_search. The old nc_ alias is
// gone from both sides; a stale override would silently retrieve nothing, so it
// stops the app here rather than at the first search.
var SEARCH_TOOL_NAME = process.env.CLINICAL_ASSISTANT_SEARCH_TOOL || 'clinical_semantic_search';
if (SEARCH_TOOL_NAME !== 'clinical_semantic_search') {
  throw new Error('CLINICAL_ASSISTANT_SEARCH_TOOL must be clinical_semantic_search (the nc_semantic_search alias was removed)');
}
var MCP_INITIALIZE_TIMEOUT_MS = positiveInt(process.env.CLINICAL_ASSISTANT_MCP_INITIALIZE_TIMEOUT_MS, 30000);
var MCP_REQUEST_TIMEOUT_MS = positiveInt(process.env.CLINICAL_ASSISTANT_MCP_REQUEST_TIMEOUT_MS, 90000);
var MCP_SESSION_TTL_MS = positiveInt(process.env.CLINICAL_ASSISTANT_MCP_SESSION_TTL_MS, 10 * 60 * 1000);
var _mcpSession = null;
var _mcpSessionPromise = null;
var _mcpCallQueue = Promise.resolve();
// How many tool calls may be in flight at once. They used to run one at a
// time behind a promise chain, so two people asking at the same moment
// waited for each other; the server handles concurrent requests on one
// session, so a small bound is enough to stop the queueing without letting a
// burst pile onto it.
var MCP_CONCURRENCY = positiveInt(process.env.CLINICAL_ASSISTANT_MCP_CONCURRENCY, 3);
var _inFlight = 0;
var _waiting = [];
// The session is renewed before it expires, in the background, so no one
// pays for a cold initialize at the start of their question.
var _warmTimer = null;
var _mcpRequestId = 1;
var MCP_CLEANUP_TIMEOUT_MS = 5000;
var _closing = false;
var _closePromise = null;
var _ownedSessions = new Set();
var _initializeController = null;

function positiveInt(value, fallback) {
  var n = Number(value);
  return Number.isFinite(n) && n > 0 ? Math.floor(n) : fallback;
}

// Shared-library sessions can own server-side Nextcloud clients. Explicit
// teardown releases those resources promptly on replacement or shutdown.
// Best effort: failed cleanup must not replace a successful result or primary error.
function endMcpSession(session) {
  if (!session || !session.sessionId) return Promise.resolve();
  if (session.cleanup) return session.cleanup;
  session.cleanup = (async function() {
    var controller = new AbortController();
    var timer;
    try {
      await Promise.race([
        axios.delete(session.mcpUrl || _lastGoodMcpUrl, {
          headers: { 'Accept': 'application/json, text/event-stream', 'mcp-session-id': session.sessionId },
          timeout: MCP_CLEANUP_TIMEOUT_MS,
          signal: controller.signal
        }),
        new Promise(function(resolve) {
          timer = setTimeout(function() { controller.abort(); resolve(); }, MCP_CLEANUP_TIMEOUT_MS);
        })
      ]);
    } catch (e) { /* best effort; server expiry covers failed/unknown sessions */ }
    finally { clearTimeout(timer); _ownedSessions.delete(session); }
  })();
  return session.cleanup;
}

function closeMcpSession() {
  if (_closePromise) return _closePromise;
  _closing = true;
  _mcpSession = null;
  if (_initializeController) _initializeController.abort();
  // Late initialize responses are owned and unwound by initializeMcpSession.
  // This is best effort, not a guarantee of remote deletion before process exit.
  var timer;
  var pending = [_mcpSessionPromise, _mcpCallQueue].concat(Array.from(_ownedSessions, endMcpSession));
  _closePromise = Promise.race([
    Promise.allSettled(pending),
    new Promise(function(resolve) { timer = setTimeout(resolve, MCP_CLEANUP_TIMEOUT_MS); })
  ]).then(function() {}).finally(function() { clearTimeout(timer); });
  return _closePromise;
}

async function semanticSearch(query, opts) {
  opts = opts || {};
  return callMcpTool(SEARCH_TOOL_NAME, {
    query: query,
    limit: opts.limit,
    doc_types: ['file'],
    score_threshold: 0,
    fusion: 'rrf',
    include_context: opts.includeContext,
    context_chars: opts.contextChars
  });
}

async function indexedTopicSuggestions(limit) {
  return callMcpTool('nc_indexed_topic_suggestions', {
    limit: limit || 12,
    sample_size: 1500,
    doc_type: 'file'
  });
}

function warmMcpSession() {
  if (!_warmTimer && MCP_SESSION_TTL_MS > 0) {
    _warmTimer = setInterval(function() {
      if (_closing) return;
      var soon = _mcpSession && _mcpSession.expiresAt - Date.now() < MCP_SESSION_TTL_MS / 3;
      if (!_mcpSession || soon) {
        if (soon) { var old = _mcpSession; _mcpSession = null; endMcpSession(old); }
        getMcpSession().catch(function() {});
      }
    }, Math.max(30000, Math.floor(MCP_SESSION_TTL_MS / 3)));
    if (_warmTimer.unref) _warmTimer.unref();
  }
  return getMcpSession();
}

function acquireSlot() {
  if (_inFlight < MCP_CONCURRENCY) { _inFlight++; return Promise.resolve(); }
  return new Promise(function(resolve) { _waiting.push(resolve); });
}

function releaseSlot() {
  var next = _waiting.shift();
  if (next) next(); else _inFlight--;
}

async function callMcpTool(name, args) {
  if (_closing) throw new Error('MCP client is closing');
  await acquireSlot();
  var started = Date.now();
  var sessionMs = 0;
  try {
    var t = Date.now();
    await getMcpSession();
    sessionMs = Date.now() - t;
    return await callMcpToolUnlocked(name, args);
  } finally {
    releaseSlot();
    // Where a slow search spent its time: opening a session or running the
    // call. The one question worth answering when "search=6657" shows up.
    console.info('[clinical-assistant] mcp ' + name + ': session=' + sessionMs + 'ms total=' + (Date.now() - started) + 'ms in_flight=' + _inFlight);
  }
}

async function callMcpToolUnlocked(name, args) {
  var session = await getMcpSession();
  var payload = {
    jsonrpc: '2.0',
    id: nextMcpRequestId(),
    method: 'tools/call',
    params: {
      name: name,
      arguments: args
    }
  };
  var search;
  for (var attempt = 0; attempt < 2; attempt++) {
    try {
      search = await mcpRequest(payload, session.sessionId, session.mcpUrl);
      break;
    } catch (e) {
      if (!isInvalidMcpSessionError(e)) throw e;
      if (_mcpSession === session) _mcpSession = null;
      endMcpSession(session);
      if (attempt === 1) throw e;
      session = await getMcpSession();
      payload.id = nextMcpRequestId();
    }
  }
  return search.result || search;
}

function nextMcpRequestId() {
  _mcpRequestId += 1;
  if (_mcpRequestId > 1000000000) _mcpRequestId = 2;
  return _mcpRequestId;
}

async function getMcpSession() {
  if (_closing) throw new Error('MCP client is closing');
  var now = Date.now();
  if (_mcpSession && _mcpSession.sessionId && _mcpSession.expiresAt > now) return _mcpSession;
  if (_mcpSessionPromise) return _mcpSessionPromise;
  // Discard stale ownership even when the replacement fails.
  var stale = _mcpSession;
  _mcpSession = null;
  if (stale) endMcpSession(stale);
  _mcpSessionPromise = initializeMcpSession().then(function(session) {
    if (_closing) {
      return endMcpSession(session).then(function() { throw new Error('MCP client is closing'); });
    }
    _mcpSession = session;
    return session;
  }).finally(function() {
    _mcpSessionPromise = null;
  });
  return _mcpSessionPromise;
}

async function initializeMcpSession() {
  var owned = null;
  var controller = new AbortController();
  _initializeController = controller;
  var timer = setTimeout(function() { controller.abort(); }, MCP_INITIALIZE_TIMEOUT_MS);
  try {
    var session = await mcpRequest({
      jsonrpc: '2.0',
      id: nextMcpRequestId(),
      method: 'initialize',
      params: {
        protocolVersion: '2024-11-05',
        capabilities: {},
        clientInfo: { name: 'ped-ai-clinical-assistant', version: '1.0.0' }
      }
    }, null, null, function(response, url) {
      var id = response.headers && response.headers['mcp-session-id'];
      if (id) {
        owned = { sessionId: id, mcpUrl: url };
        _ownedSessions.add(owned);
        if (_closing) endMcpSession(owned);
      }
    }, controller.signal);
    if (!owned || !session.result) throw new Error('MCP initialize failed');
    if (_closing) throw new Error('MCP client is closing');
    await mcpRequest({ jsonrpc: '2.0', method: 'notifications/initialized', params: {} },
      owned.sessionId, owned.mcpUrl, null, controller.signal);
    if (_closing) throw new Error('MCP client is closing');
    owned.expiresAt = Date.now() + MCP_SESSION_TTL_MS;
    return owned;
  } catch (e) {
    await endMcpSession(owned);
    throw e;
  } finally { clearTimeout(timer); _initializeController = null; }
}

async function mcpRequest(payload, sessionId, preferredUrl, onResponse, signal) {
  var headers = {
    'Accept': 'application/json, text/event-stream',
    'Content-Type': 'application/json'
  };
  if (sessionId) headers['mcp-session-id'] = sessionId;
  // A trickling response must not hold the shared call queue indefinitely.
  signal = signal || AbortSignal.timeout(MCP_REQUEST_TIMEOUT_MS);
  try {
    var resp = await postMcpWithRetry(payload, headers, preferredUrl, onResponse, signal);
    var body = resp.data;
    // Streaming initialize exposes the session header before a body read can fail.
    if (body && typeof body[Symbol.asyncIterator] === 'function') {
      var chunks = [];
      for await (var chunk of body) chunks.push(Buffer.from(chunk));
      body = Buffer.concat(chunks).toString('utf8');
    }
    var parsed = parseMcpResponse(body);
    if (parsed.error) throw new Error(parsed.error.message || 'MCP request failed');
    parsed.sessionId = resp.headers['mcp-session-id'] || sessionId || null;
    parsed.mcpUrl = resp.config && resp.config.url ? resp.config.url : preferredUrl || _lastGoodMcpUrl;
    return parsed;
  } catch (e) {
    // Callers log messages: never propagate raw upstream bodies, credentials or IDs.
    var safe = new Error('MCP request failed');
    safe.invalidSession = isInvalidMcpSessionError(e);
    throw safe;
  }
}

async function postMcpWithRetry(payload, headers, preferredUrl, onResponse, signal) {
  var lastErr = null;
  var isNotification = payload && payload.method === 'notifications/initialized';
  if (isNotification && !preferredUrl) throw new Error('MCP notification requires its session endpoint');
  // Session handshake belongs to its initializer, not an alternate server.
  var urls = isNotification ? [preferredUrl] : orderedMcpUrls(preferredUrl);
  var isInitialize = payload && payload.method === 'initialize';
  for (var round = 0; round < 2; round++) {
    for (var i = 0; i < urls.length; i++) {
      if (_closing) throw new Error('MCP client is closing');
      var url = urls[i];
      try {
        var resp = await axios.post(url, payload, {
          headers: headers,
          timeout: isInitialize ? MCP_INITIALIZE_TIMEOUT_MS : MCP_REQUEST_TIMEOUT_MS,
          signal: signal,
          responseType: isInitialize ? 'stream' : 'text',
          transformResponse: [function(data) { return data; }]
        });
        if (onResponse) onResponse(resp, url);
        _lastGoodMcpUrl = url;
        return resp;
      } catch (e) {
        if (onResponse && e.response) {
          onResponse(e.response, url);
          if (e.response.data && typeof e.response.data.destroy === 'function') e.response.data.destroy();
        }
        lastErr = e;
        // Only connection refusal proves initialize could not have created a session.
        if (!isTransientMcpError(e) || (isInitialize && (e.code !== 'ECONNREFUSED' || e.response))) throw e;
        console.warn('[clinical-assistant] MCP endpoint unavailable');
      }
    }
    await sleep(750 * (round + 1), undefined, { signal: signal });
  }
  throw lastErr;
}

function orderedMcpUrls(preferredUrl) {
  var candidates = [];
  [preferredUrl, _lastGoodMcpUrl].concat(MCP_URLS).forEach(function(url) {
    if (url && candidates.indexOf(url) === -1) candidates.push(url);
  });
  return candidates;
}

function isTransientMcpError(e) {
  var code = e && (e.code || (e.cause && e.cause.code)) || '';
  var status = e && e.response && e.response.status;
  return code === 'ECONNREFUSED' || code === 'ECONNRESET' || code === 'ETIMEDOUT' || status === 502 || status === 503 || status === 504;
}

function isInvalidMcpSessionError(e) {
  var status = e && e.response && e.response.status;
  var msg = String(e && e.message || '');
  return !!(e && e.invalidSession) || status === 400 || status === 404 || status === 410 || /session|mcp-session-id/i.test(msg);
}

async function getMcpHealth() {
  var results = [];
  var urls = orderedMcpUrls();
  for (var i = 0; i < urls.length; i++) {
    var mcpUrl = urls[i];
    var healthUrl = mcpUrl.replace(/\/mcp\/?$/, '/health/live');
    try {
      var resp = await axios.get(healthUrl, { timeout: 3000 });
      results.push({ url: mcpUrl, healthUrl: healthUrl, ok: true, status: resp.status, data: resp.data });
      _lastGoodMcpUrl = mcpUrl;
      return { ok: true, activeUrl: mcpUrl, checked: results, candidates: MCP_URLS };
    } catch (e) {
      results.push({ url: mcpUrl, healthUrl: healthUrl, ok: false, error: e.code || e.message });
    }
  }
  return { ok: false, activeUrl: _lastGoodMcpUrl || null, checked: results, candidates: MCP_URLS };
}

function buildMcpUrls() {
  var raw = [];
  if (process.env.CLINICAL_ASSISTANT_MCP_URLS) raw = raw.concat(process.env.CLINICAL_ASSISTANT_MCP_URLS.split(','));
  raw.push(process.env.CLINICAL_ASSISTANT_MCP_URL || process.env.MCP_SERVER_URL || '');
  raw.push('http://mcp:8000/mcp');
  raw.push('http://mcp-server-mcp-1:8000/mcp');
  raw.push('http://127.0.0.1:8100/mcp');
  var out = [];
  raw.forEach(function(url) {
    url = String(url || '').trim();
    if (!url) return;
    if (!/\/mcp\/?$/.test(url)) url = url.replace(/\/$/, '') + '/mcp';
    if (out.indexOf(url) === -1) out.push(url);
  });
  return out;
}

function parseMcpResponse(body) {
  var text = String(body || '').trim();
  if (!text) return {};
  if (text.charAt(0) === '{') return JSON.parse(text);
  var lines = text.split(/\r?\n/);
  for (var i = 0; i < lines.length; i++) {
    if (lines[i].indexOf('data:') === 0) {
      return JSON.parse(lines[i].substring(5).trim());
    }
  }
  throw new Error('Unexpected MCP response format');
}

module.exports = {
  semanticSearch: semanticSearch,
  indexedTopicSuggestions: indexedTopicSuggestions,
  getMcpHealth: getMcpHealth,
  warmMcpSession: warmMcpSession,
  closeMcpSession: closeMcpSession
};