Function runLocalHandshakeProxy

src/mcp/proxy.ts:217–417

(deps: LocalHandshakeDeps)

 * never costs the old fall-back-to-direct robustness.
 */
export async function runLocalHandshakeProxy(deps: LocalHandshakeDeps): Promise<void> {
  let daemonStatus: 'connecting' | 'ready' | 'failed' = 'connecting';
  let daemonSocket: net.Socket | null = null;
  let clientInitId: unknown = undefined;   // suppress the daemon's reply to the forwarded initialize
  // Telemetry attribution for the in-process fallback only — calls routed to
  // the daemon are counted by the daemon's own session (which receives the
  // forwarded initialize, clientInfo included), never double-counted here.
  let telemetryClient: ClientInfo | undefined;
  const pending: string[] = [];            // client lines buffered until the daemon resolves
  let engine: MCPEngine | null = null;
  let engineReady: Promise<void> | null = null;
  let shuttingDown = false;
  // Requests forwarded to the daemon and not yet answered, keyed by JSON-RPC id.
  // If the daemon dies mid-session (#662 — e.g. an MCP host SIGTERM's it when a
  // new session starts), these would otherwise hang forever; we re-serve them
  // in-process so the host always gets a reply.
  const inflight = new Map<unknown, string>();
  // Explore call history for the ONE host connection this proxy serves (CG-17).
  // Only the daemon-unavailable fallback below uses it; when the daemon is up,
  // the tracking happens on the daemon's own MCPSession.
  const exploreSession = new ExploreSessionState();
  const trackInflight = (line: string): void => {
    try {
      const m = JSON.parse(line) as JsonRpc;
      if (m && m.id !== undefined && typeof m.method === 'string' && m.method !== 'initialize') {
        inflight.set(m.id, line);
      }
    } catch { /* unparseable — nothing we could re-serve anyway */ }
  };

  const writeClient = (obj: JsonRpc | string): void => {
    try { process.stdout.write((typeof obj === 'string' ? obj : JSON.stringify(obj)) + '\n'); } catch { /* host gone */ }
  };
  const shutdown = (): void => {
    if (shuttingDown) return; shuttingDown = true;
    try { daemonSocket?.destroy(); } catch { /* ignore */ }
    try { engine?.stop(); } catch { /* ignore */ }
    process.exit(0);
  };
  const ensureEngine = (): Promise<void> => {
    if (!engine) engine = deps.makeEngine();
    if (!engineReady) engineReady = engine.ensureInitialized(deps.root).catch(() => { /* degraded */ });
    return engineReady;
  };
  // Daemon-unavailable fallback: serve a client message in-process.
  const handleLocally = async (line: string): Promise<void> => {
    let msg: JsonRpc; try { msg = JSON.parse(line) as JsonRpc; } catch { return; }
    const id = msg.id;
    if (msg.method === 'tools/call' && id !== undefined) {
      try {
        await ensureEngine();
        const params = (msg.params || {}) as { name: string; arguments?: Record<string, unknown> };
        const result = await engine!.getToolHandler().execute(params.name, params.arguments || {}, exploreSession);
        writeClient({ jsonrpc: '2.0', id, result });
        getTelemetry().recordUsage('mcp_tool', params.name, !result.isError, telemetryClient);
      } catch (err) {
        writeClient({ jsonrpc: '2.0', id, error: { code: -32603, message: err instanceof Error ? err.message : String(err) } });
      }