fix(indexing): bounded-memory yielding pipeline tail + daemon session fixes (#1212) (#1226)

Large-codebase indexing died at the end of "Resolving refs" two ways:
watchdog kills of healthy work (24k-file Java on Windows, #1212 — third
iteration of the #1091/#1122 class) and hard OOMs (Linux kernel scale,
where v1.3.0 could not complete at any watchdog setting). Root causes:
~31 of 37 dynamic-edge synthesis passes ran start-to-finish with no
yield points, several materialized whole-graph snapshots (kotlin
expect/actual opened with getAllNodes() — 2M nodes in one array; the
C fn-pointer pass retained every C file's contents twice plus every
function node), and the post-index WAL checkpoint ran minutes of
synchronous IO on the main thread, killing even a successful index at
the finish line.

The pipeline tail now follows the same discipline as the rest: never
hold O(graph) in the heap, yield everywhere.

- All synthesis passes stream node-kind scans (cursors, not arrays) and
  yield on time-budgeted checkpoints; language gates skip passes whose
  filters a project's file languages provably can't satisfy.
- kotlin expect/actual filters SQL-side; c-fnptr caches are LRU-bounded,
  units stream one file at a time, and the all-functions array +
  write-only id map are gone; spring reads each .java once, not twice.
- runMaintenance moved to a worker thread (own SQLite connection);
  per-file store commits chunk with yields behind a serialized flush
  chain (preserving #1015 file-order determinism); resolver warm-up
  streams the DISTINCT name set; resolution batch-tail and merged-edge
  inserts run in bounded sub-transactions.
- Daemon: fixed a socket-handoff race that could leave a fresh MCP
  session permanently silent (client-hello tail unshifted into a
  flowing stream with zero listeners — the long-standing #662 test
  flake was this real bug); first tool call no longer queues behind
  the query pool's cold start (pool.ready gate).

Validation: Linux kernel (70,129 files, 2.05M nodes, 6.4M edges) fully
indexes in 27m8s on a 2-core/6GB container at default heap + default
watchdog; llvm-project (180k files) completes under 1GB RSS including
kill-and-sync recovery; synthesized-edge and full-graph parity are
byte-identical vs baseline on elasticsearch/redis/vim; the ex-flaky
daemon test passed 25/25 under load. Env-gated diagnostics kept:
CODEGRAPH_SYNTH_TIMINGS pass/phase timings, CODEGRAPH_MCP_DEBUG hop
tracing. Design record: docs/design/main-thread-stall-followup.md.

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

src/mcp/daemon.ts

@@ -803,10 +803,24 @@ function readClientHello(
     ) => {
       if (settled) return;
       settled = true;
+      // PAUSE before detaching: removing the last 'data' listener does NOT
+      // stop a flowing stream, so bytes arriving (or unshifted) in the gap
+      // between this handler and the session transport attaching were emitted
+      // to zero listeners and silently DISCARDED — and the listener swap left
+      // the socket's flow state wedged, never delivering to the new listener.
+      // A proxy whose client-hello arrived glued to the initialize hit this
+      // ~1-in-5 under load: the daemon answered nothing for the whole session
+      // (the #662 test flake, and real dead sessions behind it). Paused, the
+      // unshifted tail and any new bytes buffer; SocketTransport.start()
+      // resumes explicitly.
+      try { socket.pause(); } catch { /* stream already gone */ }
       socket.removeListener('data', onData);
       socket.removeListener('error', onEnd);
       socket.removeListener('close', onEnd);
       clearTimeout(timer);
+      if (process.env.CODEGRAPH_MCP_DEBUG) {
+        process.stderr.write(`[mcp-debug] clientHello finish pid=${String(peers.pid)} putBack=${putBack ? putBack.length : 0} flowing=${String(socket.readableFlowing)}\n`);
+      }
       if (putBack && putBack.length > 0 && !socket.destroyed) {
         try { socket.unshift(putBack); } catch { /* stream already gone */ }
       }
@@ -836,7 +850,12 @@ function readClientHello(
       }
     };
     const onEnd = () => finish({ pid: null, hostPid: null });
-    const timer = setTimeout(() => finish({ pid: null, hostPid: null }), CLIENT_HELLO_TIMEOUT_MS);
+    // On timeout, hand back whatever partial bytes accumulated — discarding
+    // them would tear the first message the transport parses.
+    const timer = setTimeout(() => {
+      const partial = chunks.length === 0 ? undefined : (chunks.length === 1 ? chunks[0] : Buffer.concat(chunks, total));
+      finish({ pid: null, hostPid: null }, partial);
+    }, CLIENT_HELLO_TIMEOUT_MS);
     timer.unref?.();
     socket.on('data', onData);
     socket.on('error', onEnd);

src/mcp/proxy.ts

@@ -279,10 +279,12 @@ export async function runLocalHandshakeProxy(deps: LocalHandshakeDeps): Promise<
   const routeToDaemon = (line: string): void => {
     if (daemonStatus === 'ready' && daemonSocket) {
       trackInflight(line);
+      if (process.env.CODEGRAPH_MCP_DEBUG) process.stderr.write(`[mcp-debug] proxy->daemon ${line.slice(0, 80)}\n`);
       try { daemonSocket.write(line.endsWith('\n') ? line : line + '\n'); } catch { /* close path */ }
     } else if (daemonStatus === 'failed') {
       void handleLocally(line);
     } else {
+      if (process.env.CODEGRAPH_MCP_DEBUG) process.stderr.write(`[mcp-debug] proxy-buffer(${daemonStatus}) ${line.slice(0, 80)}\n`);
       pending.push(line);
     }
   };
@@ -364,6 +366,7 @@ export async function runLocalHandshakeProxy(deps: LocalHandshakeDeps): Promise<
         if (!line.trim()) continue;
         let resp: JsonRpc | null = null;
         try { resp = JSON.parse(line) as JsonRpc; } catch { /* not JSON — relay verbatim */ }
+        if (process.env.CODEGRAPH_MCP_DEBUG) process.stderr.write(`[mcp-debug] daemon->proxy ${line.slice(0, 80)}\n`);
         if (resp && resp.id !== undefined && ('result' in resp || 'error' in resp)) {
           inflight.delete(resp.id); // answered — no longer in flight
           // Suppress the daemon's reply to the initialize we forwarded to prime it
@@ -392,7 +395,11 @@ export async function runLocalHandshakeProxy(deps: LocalHandshakeDeps): Promise<
     };
     socket.on('close', onDaemonLost);
     socket.on('error', onDaemonLost);
-    for (const line of pending) { trackInflight(line); try { socket.write(line + '\n'); } catch { /* ignore */ } }
+    for (const line of pending) {
+      trackInflight(line);
+      if (process.env.CODEGRAPH_MCP_DEBUG) process.stderr.write(`[mcp-debug] proxy-flush ${line.slice(0, 80)}\n`);
+      try { socket.write(line + '\n'); } catch { /* ignore */ }
+    }
     pending.length = 0;
   } else if (!shuttingDown) {
     daemonStatus = 'failed';

src/mcp/query-pool.ts

@@ -183,6 +183,23 @@ export class QueryPool {
     return !this.destroyed && this.totalCrashes < CRASH_BUDGET;
   }
 
+  /**
+   * True once at least one worker has completed its cold start (posted the
+   * 'ready' handshake). Until then the ToolHandler serves calls IN-PROCESS:
+   * a worker cold start is a full module load + DB open — seconds normally,
+   * tens of seconds on a loaded machine — and a call queued behind it gets
+   * nothing until the 45s busy backstop. The daemon's very first tool call
+   * hitting that window was the recurring #662 test flake (and a real
+   * first-call stall for agents). The pool exists for CONCURRENT load, which
+   * by definition arrives after warm-up; the pre-pool in-process path is
+   * strictly better while nothing is warm. Stays true for the pool's
+   * lifetime — later crash-respawn gaps are covered by retry + backstop.
+   */
+  get ready(): boolean {
+    return this.everReady && !this.destroyed;
+  }
+  private everReady = false;
+
   private spawnOne(): void {
     if (this.destroyed || this.workers.size >= this.maxSize) return;
     let w: PoolWorker;
@@ -204,6 +221,7 @@ export class QueryPool {
     if (m.type === 'ready') {
       this.pendingWorkers.delete(w);
       if (m.ok === false) this.totalCrashes++; // hard open failure
+      else this.everReady = true;
       this.idle.push(w);
       this.drain();
       return;

src/mcp/session.ts

@@ -260,9 +260,12 @@ export class MCPSession {
       return;
     }
 
+    if (process.env.CODEGRAPH_MCP_DEBUG) process.stderr.write(`[mcp-debug] toolsCall ${toolName} id=${String(request.id)} pre-init\n`);
     await this.retryInitIfNeeded();
 
+    if (process.env.CODEGRAPH_MCP_DEBUG) process.stderr.write(`[mcp-debug] toolsCall ${toolName} id=${String(request.id)} dispatch\n`);
     const result = await this.engine.getToolHandler().execute(toolName, toolArgs);
+    if (process.env.CODEGRAPH_MCP_DEBUG) process.stderr.write(`[mcp-debug] toolsCall ${toolName} id=${String(request.id)} done\n`);
     this.transport.sendResult(request.id, result);
     // After the reply is on the wire — telemetry must never delay a tool
     // response (in-memory increment only; see src/telemetry).

src/mcp/tools.ts

@@ -1399,15 +1399,20 @@ export class ToolHandler {
       }
 
       // Read tools: off-load the CPU-heavy dispatch to the worker pool when one
-      // is attached and healthy (daemon mode), so the daemon's single event loop
-      // stays free for the MCP transport under concurrent load — otherwise N
-      // concurrent explores serialize AND starve the transport until the whole
-      // batch drains (clients then time out). With no pool (direct mode) or a
-      // degraded one, dispatch runs in-process exactly as before. Either way the
-      // result flows through the cross-cutting notices — worktree-index mismatch
-      // (#155) and per-file staleness (#403) — which need the watched MAIN
-      // instance and so are always applied here, never in the worker.
-      const result = (this.queryPool && this.queryPool.healthy)
+      // is attached, healthy, AND has finished its first cold start (daemon
+      // mode), so the daemon's single event loop stays free for the MCP
+      // transport under concurrent load — otherwise N concurrent explores
+      // serialize AND starve the transport until the whole batch drains
+      // (clients then time out). Before the first worker is warm, calls run
+      // in-process: a call queued behind a cold start sat invisible until the
+      // 45s busy backstop — the daemon's first tool call stalling for however
+      // long a worker spawn takes on a loaded machine (the #662 flake). With
+      // no pool (direct mode) or a degraded one, dispatch runs in-process
+      // exactly as before. Either way the result flows through the
+      // cross-cutting notices — worktree-index mismatch (#155) and per-file
+      // staleness (#403) — which need the watched MAIN instance and so are
+      // always applied here, never in the worker.
+      const result = (this.queryPool && this.queryPool.healthy && this.queryPool.ready)
         ? await this.queryPool.run(toolName, args)
         : await this.executeReadTool(toolName, args);
       const withWorktree = this.withWorktreeNotice(result, args.projectPath as string | undefined);

src/mcp/transport.ts

@@ -193,7 +193,15 @@ abstract class LineBasedJsonRpcTransport implements JsonRpcTransport {
 
     if (this.messageHandler) {
       try {
+        if (process.env.CODEGRAPH_MCP_DEBUG) {
+          const m = parsed as { method?: string; id?: unknown };
+          process.stderr.write(`[mcp-debug] recv method=${m.method} id=${String(m.id)}\n`);
+        }
         await this.messageHandler(parsed as JsonRpcRequest | JsonRpcNotification);
+        if (process.env.CODEGRAPH_MCP_DEBUG) {
+          const m = parsed as { method?: string; id?: unknown };
+          process.stderr.write(`[mcp-debug] handled method=${m.method} id=${String(m.id)}\n`);
+        }
       } catch (err) {
         const message = parsed as JsonRpcRequest;
         if ('id' in message) {
@@ -353,7 +361,11 @@ export class SocketTransport extends LineBasedJsonRpcTransport {
     this.messageHandler = handler;
 
     this.socket.setEncoding('utf8');
+    if (process.env.CODEGRAPH_MCP_DEBUG) {
+      process.stderr.write(`[mcp-debug] transport attached flowing=${String(this.socket.readableFlowing)} buffered=${this.socket.readableLength}\n`);
+    }
     this.socket.on('data', (chunk: string) => {
+      if (process.env.CODEGRAPH_MCP_DEBUG) process.stderr.write(`[mcp-debug] transport data ${chunk.length}b\n`);
       this.buffer += chunk;
       let idx;
       // Drain every complete line; tail-fragment stays in the buffer for the
@@ -374,6 +386,11 @@ export class SocketTransport extends LineBasedJsonRpcTransport {
       process.stderr.write(`[CodeGraph daemon] socket error: ${err.message}\n`);
       this.handleSocketClose();
     });
+    // The daemon's hello reader hands the socket over PAUSED (so the unshifted
+    // tail can't be emitted to zero listeners and lost — the #662 wedge).
+    // Attaching 'data' does not resume an explicitly-paused stream; do it here.
+    // Harmless when the socket was never paused.
+    this.socket.resume();
   }
 
   stop(): void {