From b4747203d85b32f5dba3f37a34ae6f3d73de3a62 Mon Sep 17 00:00:00 2001 From: abhigyanpatwari Date: Wed, 25 Feb 2026 17:10:51 +0530 Subject: [PATCH 1/7] =?UTF-8?q?feat(gitnexus):=20v1.2.9=20=E2=80=94=20impa?= =?UTF-8?q?ct=20enrichment,=20cypher=20markdown,=20Windows=20setup=20fix?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Impact tool now returns risk score, affected processes/modules, and summary - Cypher tool formats results as markdown tables for LLM readability - Context tool includes module (functional area) field - Semantic search skips model init when embeddings are disabled - Setup: wrap npx in cmd /c on Windows for .cmd script compatibility - Embedder: silence stderr during ONNX model load to protect MCP stdio - API: use executeCypher directly to avoid double formatting - Add community integrations section to READMEs Co-Authored-By: Claude Opus 4.6 --- AGENTS.md | 2 +- CLAUDE.md | 2 +- README.md | 6 ++ gitnexus/README.md | 6 ++ gitnexus/package-lock.json | 4 +- gitnexus/package.json | 2 +- gitnexus/src/cli/setup.ts | 9 +- gitnexus/src/mcp/core/embedder.ts | 12 ++- gitnexus/src/mcp/local/local-backend.ts | 127 ++++++++++++++++++++++-- gitnexus/src/mcp/tools.ts | 15 ++- gitnexus/src/server/api.ts | 6 +- 11 files changed, 166 insertions(+), 25 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 322c9439e..d5b788b41 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -3,7 +3,7 @@ # GitNexus MCP -This project is indexed by GitNexus as **GitnexusV2** (1309 symbols, 3350 relationships, 101 execution flows). +This project is indexed by GitNexus as **GitnexusV2** (1348 symbols, 3469 relationships, 104 execution flows). GitNexus provides a knowledge graph over this codebase — call chains, blast radius, execution flows, and semantic search. diff --git a/CLAUDE.md b/CLAUDE.md index b4f97d4c0..3fbcde034 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -1,7 +1,7 @@ # GitNexus MCP -This project is indexed by GitNexus as **GitnexusV2** (1309 symbols, 3350 relationships, 101 execution flows). +This project is indexed by GitNexus as **GitnexusV2** (1348 symbols, 3469 relationships, 104 execution flows). GitNexus provides a knowledge graph over this codebase — call chains, blast radius, execution flows, and semantic search. diff --git a/README.md b/README.md index b36cd313b..2cf84826f 100644 --- a/README.md +++ b/README.md @@ -72,6 +72,12 @@ To configure MCP for your editor, run `npx gitnexus setup` once — or set it up > **Claude Code** gets the deepest integration: MCP tools + agent skills + PreToolUse hooks that automatically enrich grep/glob/bash calls with knowledge graph context. +### Community Integrations + +| Agent | Install | Source | +|-------|---------|--------| +| [pi](https://pi.dev) | `pi install npm:pi-gitnexus` | [pi-gitnexus](https://github.com/tintinweb/pi-gitnexus) | + If you prefer manual configuration: **Claude Code** (full support — MCP + skills + hooks): diff --git a/gitnexus/README.md b/gitnexus/README.md index b69cb1a10..e6aa62940 100644 --- a/gitnexus/README.md +++ b/gitnexus/README.md @@ -39,6 +39,12 @@ To configure MCP for your editor, run `npx gitnexus setup` once — or set it up > **Claude Code** gets the deepest integration: MCP tools + agent skills + PreToolUse hooks that automatically enrich grep/glob/bash calls with knowledge graph context. +### Community Integrations + +| Agent | Install | Source | +|-------|---------|--------| +| [pi](https://pi.dev) | `pi install npm:pi-gitnexus` | [pi-gitnexus](https://github.com/tintinweb/pi-gitnexus) | + ## MCP Setup (manual) If you prefer to configure manually instead of using `gitnexus setup`: diff --git a/gitnexus/package-lock.json b/gitnexus/package-lock.json index 86415b4d9..b937d11b9 100644 --- a/gitnexus/package-lock.json +++ b/gitnexus/package-lock.json @@ -1,12 +1,12 @@ { "name": "gitnexus", - "version": "1.1.9", + "version": "1.2.9", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "gitnexus", - "version": "1.1.9", + "version": "1.2.9", "license": "PolyForm-Noncommercial-1.0.0", "dependencies": { "@huggingface/transformers": "^3.0.0", diff --git a/gitnexus/package.json b/gitnexus/package.json index a4c3ea1ce..261be4bfc 100644 --- a/gitnexus/package.json +++ b/gitnexus/package.json @@ -1,6 +1,6 @@ { "name": "gitnexus", - "version": "1.2.8", + "version": "1.2.9", "description": "Graph-powered code intelligence for AI agents. Index any codebase, query via MCP or CLI.", "author": "Abhigyan Patwari", "license": "PolyForm-Noncommercial-1.0.0", diff --git a/gitnexus/src/cli/setup.ts b/gitnexus/src/cli/setup.ts index 77515a49a..a68b3ac0a 100644 --- a/gitnexus/src/cli/setup.ts +++ b/gitnexus/src/cli/setup.ts @@ -22,9 +22,16 @@ interface SetupResult { } /** - * The MCP server entry for all editors + * The MCP server entry for all editors. + * On Windows, npx must be invoked via cmd /c since it's a .cmd script. */ function getMcpEntry() { + if (process.platform === 'win32') { + return { + command: 'cmd', + args: ['/c', 'npx', '-y', 'gitnexus@latest', 'mcp'], + }; + } return { command: 'npx', args: ['-y', 'gitnexus@latest', 'mcp'], diff --git a/gitnexus/src/mcp/core/embedder.ts b/gitnexus/src/mcp/core/embedder.ts index 097b13fd7..ee480a6a9 100644 --- a/gitnexus/src/mcp/core/embedder.ts +++ b/gitnexus/src/mcp/core/embedder.ts @@ -43,10 +43,13 @@ export const initEmbedder = async (): Promise => { for (const device of devicesToTry) { try { - // Silence stdout during model load — ONNX Runtime and transformers.js - // may write progress/init messages to stdout which corrupts MCP stdio protocol. - const origWrite = process.stdout.write; + // Silence stdout and stderr during model load — ONNX Runtime and transformers.js + // may write progress/init messages that corrupt MCP stdio protocol or produce + // noisy warnings (e.g. node assignment to execution providers). + const origStdout = process.stdout.write; + const origStderr = process.stderr.write; process.stdout.write = (() => true) as any; + process.stderr.write = (() => true) as any; try { embedderInstance = await (pipeline as any)( 'feature-extraction', @@ -57,7 +60,8 @@ export const initEmbedder = async (): Promise => { } ); } finally { - process.stdout.write = origWrite; + process.stdout.write = origStdout; + process.stderr.write = origStderr; } console.error(`GitNexus: Embedding model loaded (${device})`); return embedderInstance!; diff --git a/gitnexus/src/mcp/local/local-backend.ts b/gitnexus/src/mcp/local/local-backend.ts index 9ae45ee6a..c47560d2b 100644 --- a/gitnexus/src/mcp/local/local-backend.ts +++ b/gitnexus/src/mcp/local/local-backend.ts @@ -279,8 +279,10 @@ export class LocalBackend { switch (method) { case 'query': return this.query(repo, params); - case 'cypher': - return this.cypher(repo, params); + case 'cypher': { + const raw = await this.cypher(repo, params); + return this.formatCypherAsMarkdown(raw); + } case 'context': return this.context(repo, params); case 'impact': @@ -395,16 +397,18 @@ export class LocalBackend { `); } catch { /* symbol might not be in any process */ } - // Get cluster cohesion as internal ranking signal (never exposed) + // Get cluster membership + cohesion (cohesion used as internal ranking signal) let cohesion = 0; + let module: string | undefined; try { const cohesionRows = await executeQuery(repo.id, ` MATCH (n {id: '${escaped}'})-[:CodeRelation {type: 'MEMBER_OF'}]->(c:Community) - RETURN c.cohesion AS cohesion + RETURN c.cohesion AS cohesion, c.heuristicLabel AS module LIMIT 1 `); if (cohesionRows.length > 0) { cohesion = (cohesionRows[0].cohesion ?? cohesionRows[0][0]) || 0; + module = cohesionRows[0].module ?? cohesionRows[0][1]; } } catch { /* no cluster info */ } @@ -429,6 +433,7 @@ export class LocalBackend { filePath: sym.filePath, startLine: sym.startLine, endLine: sym.endLine, + ...(module ? { module } : {}), ...(includeContent && content ? { content } : {}), }; @@ -577,6 +582,10 @@ export class LocalBackend { */ private async semanticSearch(repo: RepoHandle, query: string, limit: number): Promise { try { + // Check if embedding table exists before loading the model (avoids heavy model init when embeddings are off) + const tableCheck = await executeQuery(repo.id, `MATCH (e:CodeEmbedding) RETURN COUNT(*) AS cnt LIMIT 1`); + if (!tableCheck.length || (tableCheck[0].cnt ?? tableCheck[0][0]) === 0) return []; + const queryVec = await embedQuery(query); const dims = getEmbeddingDims(); const queryVecStr = `[${queryVec.join(',')}]`; @@ -630,8 +639,8 @@ export class LocalBackend { } return results; - } catch (err: any) { - console.error('GitNexus: Semantic search unavailable -', err.message); + } catch { + // Expected when embeddings are disabled — silently fall back to BM25-only return []; } } @@ -643,11 +652,11 @@ export class LocalBackend { private async cypher(repo: RepoHandle, params: { query: string }): Promise { await this.ensureInitialized(repo.id); - + if (!isKuzuReady(repo.id)) { return { error: 'KuzuDB not ready. Index may be corrupted.' }; } - + try { const result = await executeQuery(repo.id, params.query); return result; @@ -656,6 +665,36 @@ export class LocalBackend { } } + /** + * Format raw Cypher result rows as a markdown table for LLM readability. + * Falls back to raw result if rows aren't tabular objects. + */ + private formatCypherAsMarkdown(result: any): any { + if (!Array.isArray(result) || result.length === 0) return result; + + const firstRow = result[0]; + if (typeof firstRow !== 'object' || firstRow === null) return result; + + const keys = Object.keys(firstRow); + if (keys.length === 0) return result; + + const header = '| ' + keys.join(' | ') + ' |'; + const separator = '| ' + keys.map(() => '---').join(' | ') + ' |'; + const dataRows = result.map((row: any) => + '| ' + keys.map(k => { + const v = row[k]; + if (v === null || v === undefined) return ''; + if (typeof v === 'object') return JSON.stringify(v); + return String(v); + }).join(' | ') + ' |' + ); + + return { + markdown: [header, separator, ...dataRows].join('\n'), + row_count: result.length, + }; + } + /** * Aggregate same-named clusters: group by heuristicLabel, sum symbols, * weighted-average cohesion, filter out tiny clusters (<5 symbols). @@ -1318,7 +1357,69 @@ export class LocalBackend { if (!grouped[item.depth]) grouped[item.depth] = []; grouped[item.depth].push(item); } - + + // ── Enrichment: affected processes, modules, risk ────────────── + const directCount = (grouped[1] || []).length; + let affectedProcesses: any[] = []; + let affectedModules: any[] = []; + + if (impacted.length > 0) { + const allIds = impacted.map(i => `'${i.id.replace(/'/g, "''")}'`).join(', '); + const d1Ids = (grouped[1] || []).map((i: any) => `'${i.id.replace(/'/g, "''")}'`).join(', '); + + // Affected processes: which execution flows are broken and at which step + const [processRows, moduleRows, directModuleRows] = await Promise.all([ + executeQuery(repo.id, ` + MATCH (s)-[r:CodeRelation {type: 'STEP_IN_PROCESS'}]->(p:Process) + WHERE s.id IN [${allIds}] + RETURN p.heuristicLabel AS name, COUNT(DISTINCT s.id) AS hits, MIN(r.step) AS minStep, p.stepCount AS stepCount + ORDER BY hits DESC + LIMIT 20 + `).catch(() => []), + executeQuery(repo.id, ` + MATCH (s)-[:CodeRelation {type: 'MEMBER_OF'}]->(c:Community) + WHERE s.id IN [${allIds}] + RETURN c.heuristicLabel AS name, COUNT(DISTINCT s.id) AS hits + ORDER BY hits DESC + LIMIT 20 + `).catch(() => []), + d1Ids ? executeQuery(repo.id, ` + MATCH (s)-[:CodeRelation {type: 'MEMBER_OF'}]->(c:Community) + WHERE s.id IN [${d1Ids}] + RETURN DISTINCT c.heuristicLabel AS name + `).catch(() => []) : Promise.resolve([]), + ]); + + affectedProcesses = processRows.map((r: any) => ({ + name: r.name || r[0], + hits: r.hits || r[1], + broken_at_step: r.minStep ?? r[2], + step_count: r.stepCount ?? r[3], + })); + + const directModuleSet = new Set(directModuleRows.map((r: any) => r.name || r[0])); + affectedModules = moduleRows.map((r: any) => { + const name = r.name || r[0]; + return { + name, + hits: r.hits || r[1], + impact: directModuleSet.has(name) ? 'direct' : 'indirect', + }; + }); + } + + // Risk scoring + const processCount = affectedProcesses.length; + const moduleCount = affectedModules.length; + let risk = 'LOW'; + if (directCount >= 30 || processCount >= 5 || moduleCount >= 5 || impacted.length >= 200) { + risk = 'CRITICAL'; + } else if (directCount >= 15 || processCount >= 3 || moduleCount >= 3 || impacted.length >= 100) { + risk = 'HIGH'; + } else if (directCount >= 5 || impacted.length >= 30) { + risk = 'MEDIUM'; + } + return { target: { id: symId, @@ -1328,6 +1429,14 @@ export class LocalBackend { }, direction, impactedCount: impacted.length, + risk, + summary: { + direct: directCount, + processes_affected: processCount, + modules_affected: moduleCount, + }, + affected_processes: affectedProcesses, + affected_modules: affectedModules, byDepth: grouped, }; } diff --git a/gitnexus/src/mcp/tools.ts b/gitnexus/src/mcp/tools.ts index 20be63278..9e2fe7eed 100644 --- a/gitnexus/src/mcp/tools.ts +++ b/gitnexus/src/mcp/tools.ts @@ -49,7 +49,7 @@ AFTER THIS: Use context() on a specific symbol for 360-degree view (callers, cal Returns results grouped by process (execution flow): - processes: ranked execution flows with relevance priority -- process_symbols: all symbols in those flows with file locations +- process_symbols: all symbols in those flows with file locations and module (functional area) - definitions: standalone types/interfaces not in any process Hybrid ranking: BM25 keyword + semantic vector search, ranked by Reciprocal Rank Fusion.`, @@ -91,6 +91,8 @@ EXAMPLES: • Trace a process: MATCH (s)-[r:CodeRelation {type: 'STEP_IN_PROCESS'}]->(p:Process) WHERE p.heuristicLabel = "UserLogin" RETURN s.name, r.step ORDER BY r.step +OUTPUT: Returns { markdown, row_count } — results formatted as a Markdown table for easy reading. + TIPS: - All relationships use single CodeRelation table — filter with {type: 'CALLS'} etc. - Community = auto-detected functional area (Leiden algorithm) @@ -172,10 +174,17 @@ Each edit is tagged with confidence: { name: 'impact', description: `Analyze the blast radius of changing a code symbol. -Returns all symbols affected by modifying the target, grouped by depth with edge types and confidence. +Returns affected symbols grouped by depth, plus risk assessment, affected execution flows, and affected modules. WHEN TO USE: Before making code changes — especially refactoring, renaming, or modifying shared code. Shows what would break. -AFTER THIS: Review d=1 items (WILL BREAK). READ gitnexus://repo/{name}/processes to check affected execution flows. +AFTER THIS: Review d=1 items (WILL BREAK). Use context() on high-risk symbols. + +Output includes: +- risk: LOW / MEDIUM / HIGH / CRITICAL +- summary: direct callers, processes affected, modules affected +- affected_processes: which execution flows break and at which step +- affected_modules: which functional areas are hit (direct vs indirect) +- byDepth: all affected symbols grouped by traversal depth Depth groups: - d=1: WILL BREAK (direct callers/importers) diff --git a/gitnexus/src/server/api.ts b/gitnexus/src/server/api.ts index fbe0ebcd6..eb690a4d9 100644 --- a/gitnexus/src/server/api.ts +++ b/gitnexus/src/server/api.ts @@ -191,9 +191,9 @@ export const createServer = async (port: number) => { return; } - const result = await backend.callTool('cypher', { repo: repoName, query: cypher }); - if (result && !Array.isArray(result) && result.error) { - res.status(500).json({ error: result.error }); + const result = await backend.executeCypher(repoName, cypher); + if (result && !Array.isArray(result) && (result as any).error) { + res.status(500).json({ error: (result as any).error }); return; } res.json({ result }); From 8089b44114903e078f6497f8f9dd90a26d1bec9f Mon Sep 17 00:00:00 2001 From: abhigyanpatwari Date: Wed, 25 Feb 2026 17:11:00 +0530 Subject: [PATCH 2/7] fix(eval): import paths, patch extraction, model configs - Fix imports from eval.agents/eval.environments to relative (agents/environments) - Add hatch wheel config for correct package discovery - Extract git diff patch from container for SWE-bench submission - Use sys.executable instead of hardcoded "python" for venv compat - Upgrade claude-haiku config to 4.5 - Add minimax-m2.1 model config Co-Authored-By: Claude Opus 4.6 --- eval/analysis/analyze_results.py | 3 ++- eval/configs/models/claude-haiku.yaml | 5 ++--- eval/configs/models/minimax-m2.1.yaml | 11 +++++++++++ eval/pyproject.toml | 4 ++++ eval/run_eval.py | 13 ++++++++++--- 5 files changed, 29 insertions(+), 7 deletions(-) create mode 100644 eval/configs/models/minimax-m2.1.yaml diff --git a/eval/analysis/analyze_results.py b/eval/analysis/analyze_results.py index 69c5f5cdb..72a1fbbb7 100644 --- a/eval/analysis/analyze_results.py +++ b/eval/analysis/analyze_results.py @@ -19,6 +19,7 @@ import json import logging import os import subprocess +import sys from pathlib import Path from typing import Any @@ -164,7 +165,7 @@ def run_swebench_evaluation(results_dir: Path, run_id: str, subset: str = "lite" try: eval_output = results_dir / run_id / "swebench_eval" cmd = [ - "python", "-m", "swebench.harness.run_evaluation", + sys.executable, "-m", "swebench.harness.run_evaluation", "--dataset_name", dataset_mapping.get(subset, subset), "--predictions_path", str(preds_path), "--max_workers", "4", diff --git a/eval/configs/models/claude-haiku.yaml b/eval/configs/models/claude-haiku.yaml index 89746cdbd..548cc7f84 100644 --- a/eval/configs/models/claude-haiku.yaml +++ b/eval/configs/models/claude-haiku.yaml @@ -1,8 +1,7 @@ -# Claude 3.5 Haiku — fast, cheap, good baseline +# Claude Haiku 4.5 — fast, cheap, good baseline # Via OpenRouter (set OPENROUTER_API_KEY in .env) -# To use Anthropic directly, change to: anthropic/claude-3-5-haiku-20241022 model: - model_name: "openrouter/anthropic/claude-3.5-haiku" + model_name: "openrouter/anthropic/claude-haiku-4.5" cost_tracking: "ignore_errors" model_kwargs: max_tokens: 8192 diff --git a/eval/configs/models/minimax-m2.1.yaml b/eval/configs/models/minimax-m2.1.yaml new file mode 100644 index 000000000..766d75a00 --- /dev/null +++ b/eval/configs/models/minimax-m2.1.yaml @@ -0,0 +1,11 @@ +# MiniMax M2.5 — via OpenRouter (set OPENROUTER_API_KEY in .env) +# Uses text-based model class because MiniMax doesn't support tool_calls natively. +# The action_regex tells mini-swe-agent to parse ```bash blocks from responses. +model: + model_class: litellm_textbased + model_name: "openrouter/minimax/minimax-m2.5" + action_regex: "```(?:bash|mswea_bash_command)\\s*\\n(.*?)\\n```" + cost_tracking: "ignore_errors" + model_kwargs: + max_tokens: 8192 + temperature: 0 diff --git a/eval/pyproject.toml b/eval/pyproject.toml index ae9d2ad92..83ccb9416 100644 --- a/eval/pyproject.toml +++ b/eval/pyproject.toml @@ -30,6 +30,10 @@ gitnexus-eval-analyze = "analysis.analyze_results:app" requires = ["hatchling"] build-backend = "hatchling.build" +[tool.hatch.build.targets.wheel] +packages = ["agents", "environments", "analysis", "bridge"] +extra-files = ["run_eval.py"] + [tool.ruff] line-length = 120 target-version = "py311" diff --git a/eval/run_eval.py b/eval/run_eval.py index 38dc7a473..7d410e7e4 100644 --- a/eval/run_eval.py +++ b/eval/run_eval.py @@ -178,7 +178,7 @@ def process_instance( env_class_name = env_config.pop("environment_class", "docker") if env_class_name == "eval.environments.gitnexus_docker.GitNexusDockerEnvironment": - from eval.environments.gitnexus_docker import GitNexusDockerEnvironment + from environments.gitnexus_docker import GitNexusDockerEnvironment env_config["image"] = get_swebench_docker_image(instance) env = GitNexusDockerEnvironment(**env_config) else: @@ -189,7 +189,7 @@ def process_instance( agent_config = dict(config.get("agent", {})) agent_class_name = agent_config.pop("agent_class", "eval.agents.gitnexus_agent.GitNexusAgent") - from eval.agents.gitnexus_agent import GitNexusAgent + from agents.gitnexus_agent import GitNexusAgent traj_path = instance_dir / f"{instance_id}.traj.json" agent_config["output_path"] = traj_path agent = GitNexusAgent(model, env, **agent_config) @@ -199,11 +199,18 @@ def process_instance( info = agent.run(instance["problem_statement"]) result["exit_status"] = info.get("exit_status") - result["submission"] = info.get("submission", "") result["cost"] = agent.cost result["n_calls"] = agent.n_calls result["gitnexus_metrics"] = agent.gitnexus_metrics.to_dict() + # Extract git diff patch from the container (SWE-bench needs the model_patch) + try: + patch_output = env.execute({"command": "cd /testbed && git diff"}) + result["submission"] = patch_output.get("output", "").strip() + except Exception as patch_err: + logger.warning(f"[{run_id}] Failed to extract patch: {patch_err}") + result["submission"] = info.get("submission", "") + except Exception as e: logger.error(f"[{run_id}] Error on {instance_id}: {e}") result["exit_status"] = type(e).__name__ From 3f0c1d3cd52d0f99785af3c3d31f5c86fbb70c8d Mon Sep 17 00:00:00 2001 From: abhigyanpatwari Date: Thu, 26 Feb 2026 10:23:23 +0530 Subject: [PATCH 3/7] feat(gitnexus): large-repo optimizations, multi-language support, bug fixes Optimize ingestion pipeline for massive codebases (Linux kernel ~75K files): - Worker pool: 8 threads, 1500 sub-batch, 30s per-batch timeout - Streaming CSV generation with per-stream MaxListeners fix - FileContentCache 3000 entries, resolveCache LRU eviction (20% at 100K cap) - Leiden 60s timeout with single-community fallback - SIGINT graceful shutdown, FTS dedup flag Multi-language node support (Struct, Enum, Macro, Impl, Trait, etc.): - Add 16 multi-language types to NodeLabel union - Separate CSV writers with correct schema (no isExported) - KuzuDB schema: backtick-escape reserved words (Macro, Union, Enum) - Add all FROM/TO pairs for multi-language relationship edges Fix bugs: getKuzuStats/deleteNodesForFile backtick escaping, insertNodeToKuzu/batchInsertNodesToKuzu missing escaping + isExported, progress bar flickering and timer display Co-Authored-By: Claude Opus 4.6 --- gitnexus/src/cli/analyze.ts | 89 ++- gitnexus/src/cli/index.ts | 21 + gitnexus/src/core/graph/graph.ts | 8 +- gitnexus/src/core/graph/types.ts | 30 +- gitnexus/src/core/ingestion/call-processor.ts | 84 +-- .../src/core/ingestion/community-processor.ts | 117 ++-- .../src/core/ingestion/filesystem-walker.ts | 69 ++- .../src/core/ingestion/import-processor.ts | 48 +- .../src/core/ingestion/parsing-processor.ts | 11 +- gitnexus/src/core/ingestion/pipeline.ts | 298 +++++----- .../src/core/ingestion/process-processor.ts | 34 +- .../core/ingestion/workers/parse-worker.ts | 77 ++- .../src/core/ingestion/workers/worker-pool.ts | 74 ++- gitnexus/src/core/kuzu/csv-generator.ts | 528 ++++++++++-------- gitnexus/src/core/kuzu/kuzu-adapter.ts | 136 ++--- gitnexus/src/core/kuzu/schema.ts | 10 + gitnexus/src/types/pipeline.ts | 20 +- 17 files changed, 1032 insertions(+), 622 deletions(-) diff --git a/gitnexus/src/cli/analyze.ts b/gitnexus/src/cli/analyze.ts index a22a8dad5..d3fbf6ba7 100644 --- a/gitnexus/src/cli/analyze.ts +++ b/gitnexus/src/cli/analyze.ts @@ -5,6 +5,8 @@ */ import path from 'path'; +import { execFileSync } from 'child_process'; +import v8 from 'v8'; import cliProgress from 'cli-progress'; import { runPipelineFromRepo } from '../core/ingestion/pipeline.js'; import { initKuzu, loadGraphToKuzu, getKuzuStats, executeQuery, executeWithReusedStatement, closeKuzu, createFTSIndex, loadCachedEmbeddings } from '../core/kuzu/kuzu-adapter.js'; @@ -16,6 +18,28 @@ import { generateAIContextFiles } from './ai-context.js'; import fs from 'fs/promises'; import { registerClaudeHook } from './claude-hooks.js'; +const HEAP_MB = 8192; +const HEAP_FLAG = `--max-old-space-size=${HEAP_MB}`; + +/** Re-exec the process with an 8GB heap if we're currently below that. */ +function ensureHeap(): boolean { + const nodeOpts = process.env.NODE_OPTIONS || ''; + if (nodeOpts.includes('--max-old-space-size')) return false; + + const v8Heap = v8.getHeapStatistics().heap_size_limit; + if (v8Heap >= HEAP_MB * 1024 * 1024 * 0.9) return false; + + try { + execFileSync(process.execPath, [HEAP_FLAG, ...process.argv.slice(1)], { + stdio: 'inherit', + env: { ...process.env, NODE_OPTIONS: `${nodeOpts} ${HEAP_FLAG}`.trim() }, + }); + } catch (e: any) { + process.exitCode = e.status ?? 1; + } + return true; +} + export interface AnalyzeOptions { force?: boolean; embeddings?: boolean; @@ -44,6 +68,8 @@ export const analyzeCommand = async ( inputPath?: string, options?: AnalyzeOptions ) => { + if (ensureHeap()) return; + console.log('\n GitNexus Analyzer\n'); let repoPath: string; @@ -88,19 +114,47 @@ export const analyzeCommand = async ( bar.start(100, 0, { phase: 'Initializing...' }); + // Graceful SIGINT handling — clean up resources and exit + let aborted = false; + const sigintHandler = () => { + if (aborted) process.exit(1); // Second Ctrl-C: force exit + aborted = true; + bar.stop(); + console.log('\n Interrupted — cleaning up...'); + closeKuzu().catch(() => {}).finally(() => process.exit(130)); + }; + process.on('SIGINT', sigintHandler); + // Route all console output through bar.log() so the bar doesn't stamp itself // multiple times when other code writes to stdout/stderr mid-render. const origLog = console.log.bind(console); const origWarn = console.warn.bind(console); const origError = console.error.bind(console); - const barLog = (...args: any[]) => origLog(args.map(a => (typeof a === 'string' ? a : String(a))).join(' ')); + const barLog = (...args: any[]) => { + // Clear the bar line, print the message, then let the next bar.update redraw + process.stdout.write('\x1b[2K\r'); + origLog(args.map(a => (typeof a === 'string' ? a : String(a))).join(' ')); + }; console.log = barLog; console.warn = barLog; console.error = barLog; - // Show elapsed seconds for phases that run longer than 3s + // Track elapsed time per phase — both updateBar and the interval use the + // same format so they don't flicker against each other. let lastPhaseLabel = 'Initializing...'; let phaseStart = Date.now(); + + /** Update bar with phase label + elapsed seconds (shown after 3s). */ + const updateBar = (value: number, phaseLabel: string) => { + if (phaseLabel !== lastPhaseLabel) { lastPhaseLabel = phaseLabel; phaseStart = Date.now(); } + const elapsed = Math.round((Date.now() - phaseStart) / 1000); + const display = elapsed >= 3 ? `${phaseLabel} (${elapsed}s)` : phaseLabel; + bar.update(value, { phase: display }); + }; + + // Tick elapsed seconds for phases with infrequent progress callbacks + // (e.g. CSV streaming, FTS indexing). Uses the same display format as + // updateBar so there's no flickering. const elapsedTimer = setInterval(() => { const elapsed = Math.round((Date.now() - phaseStart) / 1000); if (elapsed >= 3) { @@ -116,7 +170,7 @@ export const analyzeCommand = async ( if (options?.embeddings && existingMeta && !options?.force) { try { - bar.update(0, { phase: 'Caching embeddings...' }); + updateBar(0, 'Caching embeddings...'); await initKuzu(kuzuPath); const cached = await loadCachedEmbeddings(); cachedEmbeddingNodeIds = cached.embeddingNodeIds; @@ -131,13 +185,11 @@ export const analyzeCommand = async ( const pipelineResult = await runPipelineFromRepo(repoPath, (progress) => { const phaseLabel = PHASE_LABELS[progress.phase] || progress.phase; const scaled = Math.round(progress.percent * 0.6); - if (phaseLabel !== lastPhaseLabel) { lastPhaseLabel = phaseLabel; phaseStart = Date.now(); } - bar.update(scaled, { phase: phaseLabel }); + updateBar(scaled, phaseLabel); }); // ── Phase 2: KuzuDB (60–85%) ────────────────────────────────────── - lastPhaseLabel = 'Loading into KuzuDB...'; phaseStart = Date.now(); - bar.update(60, { phase: lastPhaseLabel }); + updateBar(60, 'Loading into KuzuDB...'); await closeKuzu(); const kuzuFiles = [kuzuPath, `${kuzuPath}.wal`, `${kuzuPath}.lock`]; @@ -148,17 +200,16 @@ export const analyzeCommand = async ( const t0Kuzu = Date.now(); await initKuzu(kuzuPath); let kuzuMsgCount = 0; - const kuzuResult = await loadGraphToKuzu(pipelineResult.graph, pipelineResult.fileContents, storagePath, (msg) => { + const kuzuResult = await loadGraphToKuzu(pipelineResult.graph, pipelineResult.repoPath, storagePath, (msg) => { kuzuMsgCount++; const progress = Math.min(84, 60 + Math.round((kuzuMsgCount / (kuzuMsgCount + 10)) * 24)); - bar.update(progress, { phase: msg }); + updateBar(progress, msg); }); const kuzuTime = ((Date.now() - t0Kuzu) / 1000).toFixed(1); const kuzuWarnings = kuzuResult.warnings; // ── Phase 3: FTS (85–90%) ───────────────────────────────────────── - lastPhaseLabel = 'Creating search indexes...'; phaseStart = Date.now(); - bar.update(85, { phase: lastPhaseLabel }); + updateBar(85, 'Creating search indexes...'); const t0Fts = Date.now(); try { @@ -174,7 +225,7 @@ export const analyzeCommand = async ( // ── Phase 3.5: Re-insert cached embeddings ──────────────────────── if (cachedEmbeddings.length > 0) { - bar.update(88, { phase: `Restoring ${cachedEmbeddings.length} cached embeddings...` }); + updateBar(88, `Restoring ${cachedEmbeddings.length} cached embeddings...`); const EMBED_BATCH = 200; for (let i = 0; i < cachedEmbeddings.length; i += EMBED_BATCH) { const batch = cachedEmbeddings.slice(i, i + EMBED_BATCH); @@ -203,8 +254,7 @@ export const analyzeCommand = async ( } if (!embeddingSkipped) { - lastPhaseLabel = 'Loading embedding model...'; phaseStart = Date.now(); - bar.update(90, { phase: lastPhaseLabel }); + updateBar(90, 'Loading embedding model...'); const t0Emb = Date.now(); await runEmbeddingPipeline( executeQuery, @@ -212,8 +262,7 @@ export const analyzeCommand = async ( (progress) => { const scaled = 90 + Math.round((progress.percent / 100) * 8); const label = progress.phase === 'loading-model' ? 'Loading embedding model...' : `Embedding ${progress.nodesProcessed || 0}/${progress.totalNodes || '?'}`; - if (label !== lastPhaseLabel) { lastPhaseLabel = label; phaseStart = Date.now(); } - bar.update(scaled, { phase: label }); + updateBar(scaled, label); }, {}, cachedEmbeddingNodeIds.size > 0 ? cachedEmbeddingNodeIds : undefined, @@ -222,14 +271,14 @@ export const analyzeCommand = async ( } // ── Phase 5: Finalize (98–100%) ─────────────────────────────────── - bar.update(98, { phase: 'Saving metadata...' }); + updateBar(98, 'Saving metadata...'); const meta = { repoPath, lastCommit: currentCommit, indexedAt: new Date().toISOString(), stats: { - files: pipelineResult.fileContents.size, + files: pipelineResult.totalFileCount, nodes: stats.nodes, edges: stats.edges, communities: pipelineResult.communityResult?.stats.totalCommunities, @@ -254,7 +303,7 @@ export const analyzeCommand = async ( } const aiContext = await generateAIContextFiles(repoPath, storagePath, projectName, { - files: pipelineResult.fileContents.size, + files: pipelineResult.totalFileCount, nodes: stats.nodes, edges: stats.edges, communities: pipelineResult.communityResult?.stats.totalCommunities, @@ -270,6 +319,8 @@ export const analyzeCommand = async ( const totalTime = ((Date.now() - t0Global) / 1000).toFixed(1); clearInterval(elapsedTimer); + process.removeListener('SIGINT', sigintHandler); + console.log = origLog; console.warn = origWarn; console.error = origError; diff --git a/gitnexus/src/cli/index.ts b/gitnexus/src/cli/index.ts index d5e5b84e0..4624e9c65 100644 --- a/gitnexus/src/cli/index.ts +++ b/gitnexus/src/cli/index.ts @@ -1,4 +1,25 @@ #!/usr/bin/env node + +// Raise Node heap limit for large repos (e.g. Linux kernel). +// Must run before any heavy allocation. If already set by the user, respect it. +if (!process.env.NODE_OPTIONS?.includes('--max-old-space-size')) { + const execArgv = process.execArgv.join(' '); + if (!execArgv.includes('--max-old-space-size')) { + // Re-spawn with a larger heap (8 GB) + const { execFileSync } = await import('node:child_process'); + try { + execFileSync(process.execPath, ['--max-old-space-size=8192', ...process.argv.slice(1)], { + stdio: 'inherit', + env: { ...process.env, NODE_OPTIONS: `${process.env.NODE_OPTIONS || ''} --max-old-space-size=8192`.trim() }, + }); + process.exit(0); + } catch (e: any) { + // If the child exited with an error code, propagate it + process.exit(e.status ?? 1); + } + } +} + import { Command } from 'commander'; import { analyzeCommand } from './analyze.js'; import { serveCommand } from './serve.js'; diff --git a/gitnexus/src/core/graph/graph.ts b/gitnexus/src/core/graph/graph.ts index 20643d8a2..4658131cc 100644 --- a/gitnexus/src/core/graph/graph.ts +++ b/gitnexus/src/core/graph/graph.ts @@ -51,11 +51,17 @@ export const createKnowledgeGraph = (): KnowledgeGraph => { get nodes(){ return Array.from(nodeMap.values()) }, - + get relationships(){ return Array.from(relationshipMap.values()) }, + iterNodes: () => nodeMap.values(), + iterRelationships: () => relationshipMap.values(), + forEachNode(fn: (node: GraphNode) => void) { nodeMap.forEach(fn); }, + forEachRelationship(fn: (rel: GraphRelationship) => void) { relationshipMap.forEach(fn); }, + getNode: (id: string) => nodeMap.get(id), + // O(1) count getters - avoid creating arrays just for length get nodeCount() { return nodeMap.size; diff --git a/gitnexus/src/core/graph/types.ts b/gitnexus/src/core/graph/types.ts index ee37d94ea..a5b32e9c6 100644 --- a/gitnexus/src/core/graph/types.ts +++ b/gitnexus/src/core/graph/types.ts @@ -15,7 +15,24 @@ export type NodeLabel = | 'Type' | 'CodeElement' | 'Community' - | 'Process'; + | 'Process' + // Multi-language node types + | 'Struct' + | 'Macro' + | 'Typedef' + | 'Union' + | 'Namespace' + | 'Trait' + | 'Impl' + | 'TypeAlias' + | 'Const' + | 'Static' + | 'Property' + | 'Record' + | 'Delegate' + | 'Annotation' + | 'Constructor' + | 'Template'; export type NodeProperties = { @@ -77,8 +94,19 @@ export interface GraphRelationship { } export interface KnowledgeGraph { + /** Returns a full array copy — prefer iterNodes() for iteration */ nodes: GraphNode[], + /** Returns a full array copy — prefer iterRelationships() for iteration */ relationships: GraphRelationship[], + /** Zero-copy iterator over nodes */ + iterNodes: () => IterableIterator, + /** Zero-copy iterator over relationships */ + iterRelationships: () => IterableIterator, + /** Zero-copy forEach — avoids iterator protocol overhead in hot loops */ + forEachNode: (fn: (node: GraphNode) => void) => void, + forEachRelationship: (fn: (rel: GraphRelationship) => void) => void, + /** Lookup a single node by id — O(1) */ + getNode: (id: string) => GraphNode | undefined, nodeCount: number, relationshipCount: number, addNode: (node: GraphNode) => void, diff --git a/gitnexus/src/core/ingestion/call-processor.ts b/gitnexus/src/core/ingestion/call-processor.ts index b3766df2b..421c6ed0c 100644 --- a/gitnexus/src/core/ingestion/call-processor.ts +++ b/gitnexus/src/core/ingestion/call-processor.ts @@ -286,39 +286,59 @@ const resolveCallTarget = ( * Filter out common built-in functions and noise * that shouldn't be tracked as calls */ -const isBuiltInOrNoise = (name: string): boolean => { - const builtIns = new Set([ - // JavaScript/TypeScript built-ins - 'console', 'log', 'warn', 'error', 'info', 'debug', - 'setTimeout', 'setInterval', 'clearTimeout', 'clearInterval', - 'parseInt', 'parseFloat', 'isNaN', 'isFinite', - 'encodeURI', 'decodeURI', 'encodeURIComponent', 'decodeURIComponent', - 'JSON', 'parse', 'stringify', - 'Object', 'Array', 'String', 'Number', 'Boolean', 'Symbol', 'BigInt', - 'Map', 'Set', 'WeakMap', 'WeakSet', - 'Promise', 'resolve', 'reject', 'then', 'catch', 'finally', - 'Math', 'Date', 'RegExp', 'Error', - 'require', 'import', 'export', - 'fetch', 'Response', 'Request', - // React hooks and common functions - 'useState', 'useEffect', 'useCallback', 'useMemo', 'useRef', 'useContext', - 'useReducer', 'useLayoutEffect', 'useImperativeHandle', 'useDebugValue', - 'createElement', 'createContext', 'createRef', 'forwardRef', 'memo', 'lazy', - // Common array/object methods - 'map', 'filter', 'reduce', 'forEach', 'find', 'findIndex', 'some', 'every', - 'includes', 'indexOf', 'slice', 'splice', 'concat', 'join', 'split', - 'push', 'pop', 'shift', 'unshift', 'sort', 'reverse', - 'keys', 'values', 'entries', 'assign', 'freeze', 'seal', - 'hasOwnProperty', 'toString', 'valueOf', - // Python built-ins - 'print', 'len', 'range', 'str', 'int', 'float', 'list', 'dict', 'set', 'tuple', - 'open', 'read', 'write', 'close', 'append', 'extend', 'update', - 'super', 'type', 'isinstance', 'issubclass', 'getattr', 'setattr', 'hasattr', - 'enumerate', 'zip', 'sorted', 'reversed', 'min', 'max', 'sum', 'abs', - ]); +/** Pre-built set (module-level singleton) to avoid re-creating per call */ +const BUILT_IN_NAMES = new Set([ + // JavaScript/TypeScript built-ins + 'console', 'log', 'warn', 'error', 'info', 'debug', + 'setTimeout', 'setInterval', 'clearTimeout', 'clearInterval', + 'parseInt', 'parseFloat', 'isNaN', 'isFinite', + 'encodeURI', 'decodeURI', 'encodeURIComponent', 'decodeURIComponent', + 'JSON', 'parse', 'stringify', + 'Object', 'Array', 'String', 'Number', 'Boolean', 'Symbol', 'BigInt', + 'Map', 'Set', 'WeakMap', 'WeakSet', + 'Promise', 'resolve', 'reject', 'then', 'catch', 'finally', + 'Math', 'Date', 'RegExp', 'Error', + 'require', 'import', 'export', + 'fetch', 'Response', 'Request', + // React hooks and common functions + 'useState', 'useEffect', 'useCallback', 'useMemo', 'useRef', 'useContext', + 'useReducer', 'useLayoutEffect', 'useImperativeHandle', 'useDebugValue', + 'createElement', 'createContext', 'createRef', 'forwardRef', 'memo', 'lazy', + // Common array/object methods + 'map', 'filter', 'reduce', 'forEach', 'find', 'findIndex', 'some', 'every', + 'includes', 'indexOf', 'slice', 'splice', 'concat', 'join', 'split', + 'push', 'pop', 'shift', 'unshift', 'sort', 'reverse', + 'keys', 'values', 'entries', 'assign', 'freeze', 'seal', + 'hasOwnProperty', 'toString', 'valueOf', + // Python built-ins + 'print', 'len', 'range', 'str', 'int', 'float', 'list', 'dict', 'set', 'tuple', + 'open', 'read', 'write', 'close', 'append', 'extend', 'update', + 'super', 'type', 'isinstance', 'issubclass', 'getattr', 'setattr', 'hasattr', + 'enumerate', 'zip', 'sorted', 'reversed', 'min', 'max', 'sum', 'abs', + // C/C++ standard library and common kernel helpers + 'printf', 'fprintf', 'sprintf', 'snprintf', 'vprintf', 'vfprintf', 'vsprintf', 'vsnprintf', + 'scanf', 'fscanf', 'sscanf', + 'malloc', 'calloc', 'realloc', 'free', 'memcpy', 'memmove', 'memset', 'memcmp', + 'strlen', 'strcpy', 'strncpy', 'strcat', 'strncat', 'strcmp', 'strncmp', 'strstr', 'strchr', 'strrchr', + 'atoi', 'atol', 'atof', 'strtol', 'strtoul', 'strtoll', 'strtoull', 'strtod', + 'sizeof', 'offsetof', 'typeof', + 'assert', 'abort', 'exit', '_exit', + 'fopen', 'fclose', 'fread', 'fwrite', 'fseek', 'ftell', 'rewind', 'fflush', 'fgets', 'fputs', + // Linux kernel common macros/helpers (not real call targets) + 'likely', 'unlikely', 'BUG', 'BUG_ON', 'WARN', 'WARN_ON', 'WARN_ONCE', + 'IS_ERR', 'PTR_ERR', 'ERR_PTR', 'IS_ERR_OR_NULL', + 'ARRAY_SIZE', 'container_of', 'list_for_each_entry', 'list_for_each_entry_safe', + 'min', 'max', 'clamp', 'abs', 'swap', + 'pr_info', 'pr_warn', 'pr_err', 'pr_debug', 'pr_notice', 'pr_crit', 'pr_emerg', + 'printk', 'dev_info', 'dev_warn', 'dev_err', 'dev_dbg', + 'GFP_KERNEL', 'GFP_ATOMIC', + 'spin_lock', 'spin_unlock', 'spin_lock_irqsave', 'spin_unlock_irqrestore', + 'mutex_lock', 'mutex_unlock', 'mutex_init', + 'kfree', 'kmalloc', 'kzalloc', 'kcalloc', 'krealloc', 'kvmalloc', 'kvfree', + 'get', 'put', +]); - return builtIns.has(name); -}; +const isBuiltInOrNoise = (name: string): boolean => BUILT_IN_NAMES.has(name); /** * Fast path: resolve pre-extracted call sites from workers. diff --git a/gitnexus/src/core/ingestion/community-processor.ts b/gitnexus/src/core/ingestion/community-processor.ts index 369715b66..2cafb9469 100644 --- a/gitnexus/src/core/ingestion/community-processor.ts +++ b/gitnexus/src/core/ingestion/community-processor.ts @@ -90,12 +90,18 @@ export const processCommunities = async ( ): Promise => { onProgress?.('Building graph for community detection...', 0); - // Step 1: Build a graphology graph from the knowledge graph - // We only include symbol nodes (Function, Class, Method) and CALLS edges - const graph = buildGraphologyGraph(knowledgeGraph); - + // Pre-check total symbol count to determine large-graph mode before building + let symbolCount = 0; + knowledgeGraph.forEachNode(node => { + if (node.label === 'Function' || node.label === 'Class' || node.label === 'Method' || node.label === 'Interface') { + symbolCount++; + } + }); + const isLarge = symbolCount > 10_000; + + const graph = buildGraphologyGraph(knowledgeGraph, isLarge); + if (graph.order === 0) { - // No nodes to cluster return { communities: [], memberships: [], @@ -103,13 +109,37 @@ export const processCommunities = async ( }; } - onProgress?.(`Running Leiden algorithm on ${graph.order} nodes...`, 30); + const nodeCount = graph.order; + const edgeCount = graph.size; - // Step 2: Run Leiden algorithm for community detection - const details = (leiden as any).detailed(graph, { - resolution: 1.0, // Default resolution, can be tuned - randomWalk: true, - }); + onProgress?.(`Running Leiden on ${nodeCount} nodes, ${edgeCount} edges${isLarge ? ` (filtered from ${symbolCount} symbols)` : ''}...`, 30); + + // Large graphs: higher resolution + capped iterations (matching Python leidenalg default of 2). + // The first 2 iterations capture ~95%+ of modularity; additional iterations have diminishing returns. + // Timeout: abort after 60s for pathological graph structures. + const LEIDEN_TIMEOUT_MS = 60_000; + let details: any; + try { + details = await Promise.race([ + Promise.resolve((leiden as any).detailed(graph, { + resolution: isLarge ? 2.0 : 1.0, + maxIterations: isLarge ? 3 : 0, + })), + new Promise((_, reject) => + setTimeout(() => reject(new Error('Leiden timeout')), LEIDEN_TIMEOUT_MS) + ), + ]); + } catch (e: any) { + if (e.message === 'Leiden timeout') { + onProgress?.('Community detection timed out, using fallback...', 60); + // Fallback: assign all nodes to community 0 + const communities: Record = {}; + graph.forEachNode((node: string) => { communities[node] = 0; }); + details = { communities, count: 1, modularity: 0 }; + } else { + throw e; + } + } onProgress?.(`Found ${details.count} communities...`, 60); @@ -150,46 +180,49 @@ export const processCommunities = async ( // ============================================================================ /** - * Build a graphology graph containing only symbol nodes and CALLS edges - * This is what the Leiden algorithm will cluster + * Build a graphology graph containing only symbol nodes and clustering edges. + * For large graphs (>10K symbols), filter out low-confidence fuzzy-global edges + * and degree-1 nodes that add noise and massively increase Leiden runtime. */ -const buildGraphologyGraph = (knowledgeGraph: KnowledgeGraph): any => { - // Use undirected graph for Leiden - it looks at edge density, not direction +const MIN_CONFIDENCE_LARGE = 0.5; + +const buildGraphologyGraph = (knowledgeGraph: KnowledgeGraph, isLarge: boolean): any => { const graph = new (Graph as any)({ type: 'undirected', allowSelfLoops: false }); - // Symbol types that should be clustered const symbolTypes = new Set(['Function', 'Class', 'Method', 'Interface']); - - // First pass: collect which nodes participate in clustering edges const clusteringRelTypes = new Set(['CALLS', 'EXTENDS', 'IMPLEMENTS']); const connectedNodes = new Set(); + const nodeDegree = new Map(); - knowledgeGraph.relationships.forEach(rel => { - if (clusteringRelTypes.has(rel.type) && rel.sourceId !== rel.targetId) { - connectedNodes.add(rel.sourceId); - connectedNodes.add(rel.targetId); - } + knowledgeGraph.forEachRelationship(rel => { + if (!clusteringRelTypes.has(rel.type) || rel.sourceId === rel.targetId) return; + if (isLarge && rel.confidence < MIN_CONFIDENCE_LARGE) return; + + connectedNodes.add(rel.sourceId); + connectedNodes.add(rel.targetId); + nodeDegree.set(rel.sourceId, (nodeDegree.get(rel.sourceId) || 0) + 1); + nodeDegree.set(rel.targetId, (nodeDegree.get(rel.targetId) || 0) + 1); }); - // Only add nodes that have at least one clustering edge - // Isolated nodes would just become singletons (skipped anyway) - knowledgeGraph.nodes.forEach(node => { - if (symbolTypes.has(node.label) && connectedNodes.has(node.id)) { - graph.addNode(node.id, { - name: node.properties.name, - filePath: node.properties.filePath, - type: node.label, - }); - } + knowledgeGraph.forEachNode(node => { + if (!symbolTypes.has(node.label) || !connectedNodes.has(node.id)) return; + // For large graphs, skip degree-1 nodes — they just become singletons or + // get absorbed into their single neighbor's community, but cost iteration time. + if (isLarge && (nodeDegree.get(node.id) || 0) < 2) return; + + graph.addNode(node.id, { + name: node.properties.name, + filePath: node.properties.filePath, + type: node.label, + }); }); - // Add edges - knowledgeGraph.relationships.forEach(rel => { - if (clusteringRelTypes.has(rel.type)) { - if (graph.hasNode(rel.sourceId) && graph.hasNode(rel.targetId) && rel.sourceId !== rel.targetId) { - if (!graph.hasEdge(rel.sourceId, rel.targetId)) { - graph.addEdge(rel.sourceId, rel.targetId); - } + knowledgeGraph.forEachRelationship(rel => { + if (!clusteringRelTypes.has(rel.type)) return; + if (isLarge && rel.confidence < MIN_CONFIDENCE_LARGE) return; + if (graph.hasNode(rel.sourceId) && graph.hasNode(rel.targetId) && rel.sourceId !== rel.targetId) { + if (!graph.hasEdge(rel.sourceId, rel.targetId)) { + graph.addEdge(rel.sourceId, rel.targetId); } } }); @@ -222,11 +255,11 @@ const createCommunityNodes = ( // Build node lookup for file paths const nodePathMap = new Map(); - knowledgeGraph.nodes.forEach(node => { + for (const node of knowledgeGraph.iterNodes()) { if (node.properties.filePath) { nodePathMap.set(node.id, node.properties.filePath); } - }); + } // Create community nodes - SKIP SINGLETONS (isolated nodes) const communityNodes: CommunityNode[] = []; diff --git a/gitnexus/src/core/ingestion/filesystem-walker.ts b/gitnexus/src/core/ingestion/filesystem-walker.ts index c7a2e5d47..b17403238 100644 --- a/gitnexus/src/core/ingestion/filesystem-walker.ts +++ b/gitnexus/src/core/ingestion/filesystem-walker.ts @@ -8,15 +8,30 @@ export interface FileEntry { content: string; } +/** Lightweight entry — path + size from stat, no content in memory */ +export interface ScannedFile { + path: string; + size: number; +} + +/** Path-only reference (for type signatures) */ +export interface FilePath { + path: string; +} + const READ_CONCURRENCY = 32; /** Skip files larger than 512KB — they're usually generated/vendored and crash tree-sitter */ const MAX_FILE_SIZE = 512 * 1024; -export const walkRepository = async ( +/** + * Phase 1: Scan repository — stat files to get paths + sizes, no content loaded. + * Memory: ~10MB for 100K files vs ~1GB+ with content. + */ +export const walkRepositoryPaths = async ( repoPath: string, onProgress?: (current: number, total: number, filePath: string) => void -): Promise => { +): Promise => { const files = await glob('**/*', { cwd: repoPath, nodir: true, @@ -24,7 +39,7 @@ export const walkRepository = async ( }); const filtered = files.filter(file => !shouldIgnorePath(file)); - const entries: FileEntry[] = []; + const entries: ScannedFile[] = []; let processed = 0; let skippedLarge = 0; @@ -38,8 +53,7 @@ export const walkRepository = async ( skippedLarge++; return null; } - const content = await fs.readFile(fullPath, 'utf-8'); - return { path: relativePath.replace(/\\/g, '/'), content }; + return { path: relativePath.replace(/\\/g, '/'), size: stat.size }; }) ); @@ -60,3 +74,48 @@ export const walkRepository = async ( return entries; }; + +/** + * Phase 2: Read file contents for a specific set of relative paths. + * Returns a Map for O(1) lookup. Silently skips files that fail to read. + */ +export const readFileContents = async ( + repoPath: string, + relativePaths: string[], +): Promise> => { + const contents = new Map(); + + for (let start = 0; start < relativePaths.length; start += READ_CONCURRENCY) { + const batch = relativePaths.slice(start, start + READ_CONCURRENCY); + const results = await Promise.allSettled( + batch.map(async relativePath => { + const fullPath = path.join(repoPath, relativePath); + const content = await fs.readFile(fullPath, 'utf-8'); + return { path: relativePath, content }; + }) + ); + + for (const result of results) { + if (result.status === 'fulfilled') { + contents.set(result.value.path, result.value.content); + } + } + } + + return contents; +}; + +/** + * Legacy API — scans and reads everything into memory. + * Used by sequential fallback path only. + */ +export const walkRepository = async ( + repoPath: string, + onProgress?: (current: number, total: number, filePath: string) => void +): Promise => { + const scanned = await walkRepositoryPaths(repoPath, onProgress); + const contents = await readFileContents(repoPath, scanned.map(f => f.path)); + return scanned + .filter(f => contents.has(f.path)) + .map(f => ({ path: f.path, content: contents.get(f.path)! })); +}; diff --git a/gitnexus/src/core/ingestion/import-processor.ts b/gitnexus/src/core/ingestion/import-processor.ts index 54b5ea070..1040201a0 100644 --- a/gitnexus/src/core/ingestion/import-processor.ts +++ b/gitnexus/src/core/ingestion/import-processor.ts @@ -18,6 +18,27 @@ export type ImportMap = Map>; export const createImportMap = (): ImportMap => new Map(); +/** Pre-built lookup structures for import resolution. Build once, reuse across chunks. */ +export interface ImportResolutionContext { + allFilePaths: Set; + allFileList: string[]; + normalizedFileList: string[]; + suffixIndex: SuffixIndex; + resolveCache: Map; +} + +/** Max entries in the resolve cache. Beyond this, the cache is cleared to bound memory. + * 100K entries ≈ 15MB — covers the most common import patterns. */ +const RESOLVE_CACHE_CAP = 100_000; + +export function buildImportResolutionContext(allPaths: string[]): ImportResolutionContext { + const allFileList = allPaths; + const normalizedFileList = allFileList.map(p => p.replace(/\\/g, '/')); + const allFilePaths = new Set(allFileList); + const suffixIndex = buildSuffixIndex(normalizedFileList, allFileList); + return { allFilePaths, allFileList, normalizedFileList, suffixIndex, resolveCache: new Map() }; +} + // ============================================================================ // LANGUAGE-SPECIFIC CONFIG // ============================================================================ @@ -276,6 +297,15 @@ const resolveImportPath = ( if (resolveCache.has(cacheKey)) return resolveCache.get(cacheKey) ?? null; const cache = (result: string | null): string | null => { + // Evict oldest 20% when cap is reached instead of clearing all + if (resolveCache.size >= RESOLVE_CACHE_CAP) { + const evictCount = Math.floor(RESOLVE_CACHE_CAP * 0.2); + const iter = resolveCache.keys(); + for (let i = 0; i < evictCount; i++) { + const key = iter.next().value; + if (key !== undefined) resolveCache.delete(key); + } + } resolveCache.set(cacheKey, result); return result; }; @@ -562,12 +592,13 @@ export const processImports = async ( importMap: ImportMap, onProgress?: (current: number, total: number) => void, repoRoot?: string, + allPaths?: string[], ) => { - // Create a Set of all file paths for fast lookup during resolution - const allFilePaths = new Set(files.map(f => f.path)); + // Use allPaths (full repo) when available for cross-chunk resolution, else fall back to chunk files + const allFileList = allPaths ?? files.map(f => f.path); + const allFilePaths = new Set(allFileList); const parser = await loadParser(); const resolveCache = new Map(); - const allFileList = files.map(f => f.path); // Pre-compute normalized file list once (forward slashes) const normalizedFileList = allFileList.map(p => p.replace(/\\/g, '/')); // Build suffix index for O(1) lookups @@ -738,18 +769,15 @@ export const processImports = async ( export const processImportsFromExtracted = async ( graph: KnowledgeGraph, - files: { path: string; content: string }[], + files: { path: string }[], extractedImports: ExtractedImport[], importMap: ImportMap, onProgress?: (current: number, total: number) => void, repoRoot?: string, + prebuiltCtx?: ImportResolutionContext, ) => { - const allFilePaths = new Set(files.map(f => f.path)); - const resolveCache = new Map(); - const allFileList = files.map(f => f.path); - const normalizedFileList = allFileList.map(p => p.replace(/\\/g, '/')); - // Build suffix index for O(1) lookups - const index = buildSuffixIndex(normalizedFileList, allFileList); + const ctx = prebuiltCtx ?? buildImportResolutionContext(files.map(f => f.path)); + const { allFilePaths, allFileList, normalizedFileList, suffixIndex: index, resolveCache } = ctx; let totalImportsFound = 0; let totalImportsResolved = 0; diff --git a/gitnexus/src/core/ingestion/parsing-processor.ts b/gitnexus/src/core/ingestion/parsing-processor.ts index e7bb344a4..c15d39a17 100644 --- a/gitnexus/src/core/ingestion/parsing-processor.ts +++ b/gitnexus/src/core/ingestion/parsing-processor.ts @@ -129,28 +129,25 @@ const processParsingWithWorkers = async ( symbolTable: SymbolTable, astCache: ASTCache, workerPool: WorkerPool, - onFileProgress?: FileProgressCallback + onFileProgress?: FileProgressCallback, ): Promise => { // Filter to parseable files only const parseableFiles: ParseWorkerInput[] = []; for (const file of files) { const lang = getLanguageFromFilename(file.path); - if (lang) { - parseableFiles.push({ path: file.path, content: file.content }); - } + if (lang) parseableFiles.push({ path: file.path, content: file.content }); } if (parseableFiles.length === 0) return { imports: [], calls: [], heritage: [] }; const total = files.length; - // Dispatch to worker pool — pool handles splitting into chunks - // Workers send progress messages during parsing so the bar updates smoothly + // Dispatch to worker pool — pool handles splitting into chunks and sub-batching const chunkResults = await workerPool.dispatch( parseableFiles, (filesProcessed) => { onFileProgress?.(Math.min(filesProcessed, total), total, 'Parsing...'); - } + }, ); // Merge results from all workers into graph and symbol table diff --git a/gitnexus/src/core/ingestion/pipeline.ts b/gitnexus/src/core/ingestion/pipeline.ts index d87aebf52..bccfa79ee 100644 --- a/gitnexus/src/core/ingestion/pipeline.ts +++ b/gitnexus/src/core/ingestion/pipeline.ts @@ -1,7 +1,7 @@ import { createKnowledgeGraph } from '../graph/graph.js'; import { processStructure } from './structure-processor.js'; import { processParsing } from './parsing-processor.js'; -import { processImports, processImportsFromExtracted, createImportMap } from './import-processor.js'; +import { processImports, processImportsFromExtracted, createImportMap, buildImportResolutionContext } from './import-processor.js'; import { processCalls, processCallsFromExtracted } from './call-processor.js'; import { processHeritage, processHeritageFromExtracted } from './heritage-processor.js'; import { processCommunities } from './community-processor.js'; @@ -9,20 +9,28 @@ import { processProcesses } from './process-processor.js'; import { createSymbolTable } from './symbol-table.js'; import { createASTCache } from './ast-cache.js'; import { PipelineProgress, PipelineResult } from '../../types/pipeline.js'; -import { walkRepository } from './filesystem-walker.js'; +import { walkRepositoryPaths, readFileContents } from './filesystem-walker.js'; +import { getLanguageFromFilename } from './utils.js'; import { createWorkerPool, WorkerPool } from './workers/worker-pool.js'; const isDev = process.env.NODE_ENV === 'development'; +/** Max bytes of source content to load per parse chunk. Each chunk's source + + * parsed ASTs + extracted records + worker serialization overhead all live in + * memory simultaneously, so this must be conservative. 20MB source ≈ 200-400MB + * peak working memory per chunk after parse expansion. */ +const CHUNK_BYTE_BUDGET = 20 * 1024 * 1024; // 20MB + +/** Max AST trees to keep in LRU cache */ +const AST_CACHE_CAP = 50; + export const runPipelineFromRepo = async ( repoPath: string, onProgress: (progress: PipelineProgress) => void ): Promise => { const graph = createKnowledgeGraph(); - const fileContents = new Map(); const symbolTable = createSymbolTable(); - // AST cache sized after file scan — start with a placeholder, resize after we know file count - let astCache = createASTCache(50); + let astCache = createASTCache(AST_CACHE_CAP); const importMap = createImportMap(); const cleanup = () => { @@ -31,13 +39,14 @@ export const runPipelineFromRepo = async ( }; try { + // ── Phase 1: Scan paths only (no content read) ───────────────────── onProgress({ phase: 'extracting', percent: 0, message: 'Scanning repository...', }); - const files = await walkRepository(repoPath, (current, total, filePath) => { + const scannedFiles = await walkRepositoryPaths(repoPath, (current, total, filePath) => { const scanProgress = Math.round((current / total) * 15); onProgress({ phase: 'extracting', @@ -48,179 +57,190 @@ export const runPipelineFromRepo = async ( }); }); - files.forEach(f => fileContents.set(f.path, f.content)); - - // Resize AST cache to fit all files — avoids re-parsing in import/call/heritage phases - astCache = createASTCache(files.length); + const totalFiles = scannedFiles.length; onProgress({ phase: 'extracting', percent: 15, message: 'Repository scanned successfully', - stats: { filesProcessed: files.length, totalFiles: files.length, nodesCreated: graph.nodeCount }, + stats: { filesProcessed: totalFiles, totalFiles, nodesCreated: graph.nodeCount }, }); + // ── Phase 2: Structure (paths only — no content needed) ──────────── onProgress({ phase: 'structure', percent: 15, message: 'Analyzing project structure...', - stats: { filesProcessed: 0, totalFiles: files.length, nodesCreated: graph.nodeCount }, + stats: { filesProcessed: 0, totalFiles, nodesCreated: graph.nodeCount }, }); - const filePaths = files.map(f => f.path); - processStructure(graph, filePaths); + const allPaths = scannedFiles.map(f => f.path); + processStructure(graph, allPaths); onProgress({ phase: 'structure', - percent: 30, + percent: 20, message: 'Project structure analyzed', - stats: { filesProcessed: files.length, totalFiles: files.length, nodesCreated: graph.nodeCount }, + stats: { filesProcessed: totalFiles, totalFiles, nodesCreated: graph.nodeCount }, }); + // ── Phase 3+4: Chunked read + parse ──────────────────────────────── + // Group parseable files into byte-budget chunks so only ~20MB of source + // is in memory at a time. Each chunk is: read → parse → extract → free. + + const parseableScanned = scannedFiles.filter(f => getLanguageFromFilename(f.path)); + const totalParseable = parseableScanned.length; + + // Build byte-budget chunks + const chunks: string[][] = []; + let currentChunk: string[] = []; + let currentBytes = 0; + for (const file of parseableScanned) { + if (currentChunk.length > 0 && currentBytes + file.size > CHUNK_BYTE_BUDGET) { + chunks.push(currentChunk); + currentChunk = []; + currentBytes = 0; + } + currentChunk.push(file.path); + currentBytes += file.size; + } + if (currentChunk.length > 0) chunks.push(currentChunk); + + const numChunks = chunks.length; + + if (isDev) { + const totalMB = parseableScanned.reduce((s, f) => s + f.size, 0) / (1024 * 1024); + console.log(`📂 Scan: ${totalFiles} paths, ${totalParseable} parseable (${totalMB.toFixed(0)}MB), ${numChunks} chunks @ ${CHUNK_BYTE_BUDGET / (1024 * 1024)}MB budget`); + } + onProgress({ phase: 'parsing', - percent: 30, - message: 'Parsing code definitions...', - stats: { filesProcessed: 0, totalFiles: files.length, nodesCreated: graph.nodeCount }, + percent: 20, + message: `Parsing ${totalParseable} files in ${numChunks} chunk${numChunks !== 1 ? 's' : ''}...`, + stats: { filesProcessed: 0, totalFiles: totalParseable, nodesCreated: graph.nodeCount }, }); - // Create worker pool for parallel parsing, with graceful fallback + // Create worker pool once, reuse across chunks let workerPool: WorkerPool | undefined; try { const workerUrl = new URL('./workers/parse-worker.js', import.meta.url); workerPool = createWorkerPool(workerUrl); } catch (err) { - // Worker pool creation failed (e.g., single core) — sequential fallback + // Worker pool creation failed — sequential fallback } - let workerData: Awaited> = null; + let filesParsedSoFar = 0; + + // AST cache sized for one chunk (sequential fallback uses it for import/call/heritage) + const maxChunkFiles = chunks.reduce((max, c) => Math.max(max, c.length), 0); + astCache = createASTCache(maxChunkFiles); + + // Build import resolution context once — suffix index, file lists, resolve cache. + // Reused across all chunks to avoid rebuilding O(files × path_depth) structures. + const importCtx = buildImportResolutionContext(allPaths); + const allPathObjects = allPaths.map(p => ({ path: p })); + + // Single-pass: parse + resolve imports/calls/heritage per chunk. + // Calls/heritage use the symbol table built so far (symbols from earlier chunks + // are already registered). This trades ~5% cross-chunk resolution accuracy for + // 200-400MB less memory — critical for Linux-kernel-scale repos. + const sequentialChunkPaths: string[][] = []; + try { - workerData = await processParsing(graph, files, symbolTable, astCache, (current, total, filePath) => { - const parsingProgress = 30 + ((current / total) * 40); - onProgress({ - phase: 'parsing', - percent: Math.round(parsingProgress), - message: 'Parsing code definitions...', - detail: filePath, - stats: { filesProcessed: current, totalFiles: total, nodesCreated: graph.nodeCount }, - }); - }, workerPool); + for (let chunkIdx = 0; chunkIdx < numChunks; chunkIdx++) { + const chunkPaths = chunks[chunkIdx]; + + // Read content for this chunk only + const chunkContents = await readFileContents(repoPath, chunkPaths); + const chunkFiles = chunkPaths + .filter(p => chunkContents.has(p)) + .map(p => ({ path: p, content: chunkContents.get(p)! })); + + // Parse this chunk (workers or sequential fallback) + const chunkWorkerData = await processParsing( + graph, chunkFiles, symbolTable, astCache, + (current, _total, filePath) => { + const globalCurrent = filesParsedSoFar + current; + const parsingProgress = 20 + ((globalCurrent / totalParseable) * 62); + onProgress({ + phase: 'parsing', + percent: Math.round(parsingProgress), + message: `Parsing chunk ${chunkIdx + 1}/${numChunks}...`, + detail: filePath, + stats: { filesProcessed: globalCurrent, totalFiles: totalParseable, nodesCreated: graph.nodeCount }, + }); + }, + workerPool, + ); + + if (chunkWorkerData) { + // Imports + await processImportsFromExtracted(graph, allPathObjects, chunkWorkerData.imports, importMap, undefined, repoPath, importCtx); + // Calls — resolve immediately, then free the array + if (chunkWorkerData.calls.length > 0) { + await processCallsFromExtracted(graph, chunkWorkerData.calls, symbolTable, importMap); + } + // Heritage — resolve immediately, then free + if (chunkWorkerData.heritage.length > 0) { + await processHeritageFromExtracted(graph, chunkWorkerData.heritage, symbolTable); + } + } else { + await processImports(graph, chunkFiles, astCache, importMap, undefined, repoPath, allPaths); + sequentialChunkPaths.push(chunkPaths); + } + + filesParsedSoFar += chunkFiles.length; + + // Clear AST cache between chunks to free memory + astCache.clear(); + // chunkContents + chunkFiles + chunkWorkerData go out of scope → GC reclaims + } } finally { await workerPool?.terminate(); } - onProgress({ - phase: 'imports', - percent: 70, - message: 'Resolving imports...', - stats: { filesProcessed: 0, totalFiles: files.length, nodesCreated: graph.nodeCount }, - }); - - if (workerData) { - // Fast path: imports already extracted by workers, just resolve paths - await processImportsFromExtracted(graph, files, workerData.imports, importMap, (current, total) => { - const importProgress = 70 + ((current / total) * 12); - onProgress({ - phase: 'imports', - percent: Math.round(importProgress), - message: 'Resolving imports...', - stats: { filesProcessed: current, totalFiles: total, nodesCreated: graph.nodeCount }, - }); - }, repoPath); - } else { - // Fallback: full parse + resolve (sequential path) - await processImports(graph, files, astCache, importMap, (current, total) => { - const importProgress = 70 + ((current / total) * 12); - onProgress({ - phase: 'imports', - percent: Math.round(importProgress), - message: 'Resolving imports...', - stats: { filesProcessed: current, totalFiles: total, nodesCreated: graph.nodeCount }, - }); - }, repoPath); + // Sequential fallback chunks: re-read source for call/heritage resolution + for (const chunkPaths of sequentialChunkPaths) { + const chunkContents = await readFileContents(repoPath, chunkPaths); + const chunkFiles = chunkPaths + .filter(p => chunkContents.has(p)) + .map(p => ({ path: p, content: chunkContents.get(p)! })); + astCache = createASTCache(chunkFiles.length); + await processCalls(graph, chunkFiles, astCache, symbolTable, importMap); + await processHeritage(graph, chunkFiles, astCache, symbolTable); + astCache.clear(); } + // Free import resolution context — suffix index + resolve cache no longer needed + // (allPathObjects and importCtx hold ~94MB+ for large repos) + allPathObjects.length = 0; + importCtx.resolveCache.clear(); + (importCtx as any).suffixIndex = null; + (importCtx as any).normalizedFileList = null; + if (isDev) { - const importsCount = graph.relationships.filter(r => r.type === 'IMPORTS').length; - console.log(`📊 Pipeline: After import phase, graph has ${importsCount} IMPORTS relationships (total: ${graph.relationshipCount})`); - } - - onProgress({ - phase: 'calls', - percent: 82, - message: 'Tracing function calls...', - stats: { filesProcessed: 0, totalFiles: files.length, nodesCreated: graph.nodeCount }, - }); - - if (workerData) { - // Fast path: calls already extracted by workers, just resolve targets - await processCallsFromExtracted(graph, workerData.calls, symbolTable, importMap, (current, total) => { - const callProgress = 82 + ((current / total) * 10); - onProgress({ - phase: 'calls', - percent: Math.round(callProgress), - message: 'Tracing function calls...', - stats: { filesProcessed: current, totalFiles: total, nodesCreated: graph.nodeCount }, - }); - }); - } else { - // Fallback: full parse + resolve (sequential path) - await processCalls(graph, files, astCache, symbolTable, importMap, (current, total) => { - const callProgress = 82 + ((current / total) * 10); - onProgress({ - phase: 'calls', - percent: Math.round(callProgress), - message: 'Tracing function calls...', - stats: { filesProcessed: current, totalFiles: total, nodesCreated: graph.nodeCount }, - }); - }); - } - - onProgress({ - phase: 'heritage', - percent: 92, - message: 'Extracting class inheritance...', - stats: { filesProcessed: 0, totalFiles: files.length, nodesCreated: graph.nodeCount }, - }); - - if (workerData) { - // Fast path: heritage already extracted by workers, just resolve symbols - await processHeritageFromExtracted(graph, workerData.heritage, symbolTable, (current, total) => { - const heritageProgress = 88 + ((current / total) * 4); - onProgress({ - phase: 'heritage', - percent: Math.round(heritageProgress), - message: 'Extracting class inheritance...', - stats: { filesProcessed: current, totalFiles: total, nodesCreated: graph.nodeCount }, - }); - }); - } else { - // Fallback: full parse + resolve (sequential path) - await processHeritage(graph, files, astCache, symbolTable, (current, total) => { - const heritageProgress = 88 + ((current / total) * 4); - onProgress({ - phase: 'heritage', - percent: Math.round(heritageProgress), - message: 'Extracting class inheritance...', - stats: { filesProcessed: current, totalFiles: total, nodesCreated: graph.nodeCount }, - }); - }); + let importsCount = 0; + for (const r of graph.iterRelationships()) { + if (r.type === 'IMPORTS') importsCount++; + } + console.log(`📊 Pipeline: graph has ${importsCount} IMPORTS, ${graph.relationshipCount} total relationships`); } + // ── Phase 5: Communities ─────────────────────────────────────────── onProgress({ phase: 'communities', - percent: 92, + percent: 82, message: 'Detecting code communities...', - stats: { filesProcessed: files.length, totalFiles: files.length, nodesCreated: graph.nodeCount }, + stats: { filesProcessed: totalFiles, totalFiles, nodesCreated: graph.nodeCount }, }); const communityResult = await processCommunities(graph, (message, progress) => { - const communityProgress = 92 + (progress * 0.06); + const communityProgress = 82 + (progress * 0.10); onProgress({ phase: 'communities', percent: Math.round(communityProgress), message, - stats: { filesProcessed: files.length, totalFiles: files.length, nodesCreated: graph.nodeCount }, + stats: { filesProcessed: totalFiles, totalFiles, nodesCreated: graph.nodeCount }, }); }); @@ -253,27 +273,28 @@ export const runPipelineFromRepo = async ( }); }); + // ── Phase 6: Processes ───────────────────────────────────────────── onProgress({ phase: 'processes', - percent: 98, + percent: 94, message: 'Detecting execution flows...', - stats: { filesProcessed: files.length, totalFiles: files.length, nodesCreated: graph.nodeCount }, + stats: { filesProcessed: totalFiles, totalFiles, nodesCreated: graph.nodeCount }, }); - // Dynamic process cap based on codebase size - const symbolCount = graph.nodes.filter(n => n.label !== 'File').length; + let symbolCount = 0; + graph.forEachNode(n => { if (n.label !== 'File') symbolCount++; }); const dynamicMaxProcesses = Math.max(20, Math.min(300, Math.round(symbolCount / 10))); const processResult = await processProcesses( graph, communityResult.memberships, (message, progress) => { - const processProgress = 98 + (progress * 0.01); + const processProgress = 94 + (progress * 0.05); onProgress({ phase: 'processes', percent: Math.round(processProgress), message, - stats: { filesProcessed: files.length, totalFiles: files.length, nodesCreated: graph.nodeCount }, + stats: { filesProcessed: totalFiles, totalFiles, nodesCreated: graph.nodeCount }, }); }, { maxProcesses: dynamicMaxProcesses, minSteps: 3 } @@ -317,18 +338,17 @@ export const runPipelineFromRepo = async ( percent: 100, message: `Graph complete! ${communityResult.stats.totalCommunities} communities, ${processResult.stats.totalProcesses} processes detected.`, stats: { - filesProcessed: files.length, - totalFiles: files.length, + filesProcessed: totalFiles, + totalFiles, nodesCreated: graph.nodeCount }, }); astCache.clear(); - return { graph, fileContents, communityResult, processResult }; + return { graph, repoPath, totalFileCount: totalFiles, communityResult, processResult }; } catch (error) { cleanup(); throw error; } }; - diff --git a/gitnexus/src/core/ingestion/process-processor.ts b/gitnexus/src/core/ingestion/process-processor.ts index 10d3261fb..587aa3359 100644 --- a/gitnexus/src/core/ingestion/process-processor.ts +++ b/gitnexus/src/core/ingestion/process-processor.ts @@ -93,7 +93,7 @@ export const processProcesses = async ( const callsEdges = buildCallsGraph(knowledgeGraph); const reverseCallsEdges = buildReverseCallsGraph(knowledgeGraph); const nodeMap = new Map(); - knowledgeGraph.nodes.forEach(n => nodeMap.set(n.id, n)); + for (const n of knowledgeGraph.iterNodes()) nodeMap.set(n.id, n); // Step 1: Find entry points (functions that call others but have few callers) const entryPoints = findEntryPoints(knowledgeGraph, reverseCallsEdges, callsEdges); @@ -221,29 +221,29 @@ const MIN_TRACE_CONFIDENCE = 0.5; const buildCallsGraph = (graph: KnowledgeGraph): AdjacencyList => { const adj = new Map(); - graph.relationships.forEach(rel => { + for (const rel of graph.iterRelationships()) { if (rel.type === 'CALLS' && rel.confidence >= MIN_TRACE_CONFIDENCE) { if (!adj.has(rel.sourceId)) { adj.set(rel.sourceId, []); } adj.get(rel.sourceId)!.push(rel.targetId); } - }); - + } + return adj; }; const buildReverseCallsGraph = (graph: KnowledgeGraph): AdjacencyList => { const adj = new Map(); - - graph.relationships.forEach(rel => { + + for (const rel of graph.iterRelationships()) { if (rel.type === 'CALLS' && rel.confidence >= MIN_TRACE_CONFIDENCE) { if (!adj.has(rel.targetId)) { adj.set(rel.targetId, []); } adj.get(rel.targetId)!.push(rel.sourceId); } - }); + } return adj; }; @@ -270,20 +270,20 @@ const findEntryPoints = ( reasons: string[]; }[] = []; - graph.nodes.forEach(node => { - if (!symbolTypes.has(node.label)) return; + for (const node of graph.iterNodes()) { + if (!symbolTypes.has(node.label)) continue; const filePath = node.properties.filePath || ''; // Skip test files entirely - if (isTestFile(filePath)) return; - + if (isTestFile(filePath)) continue; + const callers = reverseCallsEdges.get(node.id) || []; const callees = callsEdges.get(node.id) || []; - + // Must have at least 1 outgoing call to trace forward - if (callees.length === 0) return; - + if (callees.length === 0) continue; + // Calculate entry point score using new scoring system const { score, reasons } = calculateEntryPointScore( node.properties.name, @@ -293,11 +293,11 @@ const findEntryPoints = ( callees.length, filePath // Pass filePath for framework detection ); - + if (score > 0) { entryPointCandidates.push({ id: node.id, score, reasons }); } - }); + } // Sort by score descending and return top candidates const sorted = entryPointCandidates.sort((a, b) => b.score - a.score); @@ -306,7 +306,7 @@ const findEntryPoints = ( if (sorted.length > 0 && isDev) { console.log(`[Process] Top 10 entry point candidates (new scoring):`); sorted.slice(0, 10).forEach((c, i) => { - const node = graph.nodes.find(n => n.id === c.id); + const node = graph.getNode(c.id); const exported = node?.properties.isExported ? '✓' : '✗'; const shortPath = node?.properties.filePath?.split('/').slice(-2).join('/') || ''; console.log(` ${i+1}. ${node?.properties.name} [exported:${exported}] (${shortPath})`); diff --git a/gitnexus/src/core/ingestion/workers/parse-worker.ts b/gitnexus/src/core/ingestion/workers/parse-worker.ts index 31e4d02e3..2eee2c5bc 100644 --- a/gitnexus/src/core/ingestion/workers/parse-worker.ts +++ b/gitnexus/src/core/ingestion/workers/parse-worker.ts @@ -252,6 +252,7 @@ const findEnclosingFunctionId = (node: any, filePath: string): string | null => }; const BUILT_INS = new Set([ + // JavaScript/TypeScript 'console', 'log', 'warn', 'error', 'info', 'debug', 'setTimeout', 'setInterval', 'clearTimeout', 'clearInterval', 'parseInt', 'parseFloat', 'isNaN', 'isFinite', @@ -270,10 +271,32 @@ const BUILT_INS = new Set([ 'push', 'pop', 'shift', 'unshift', 'sort', 'reverse', 'keys', 'values', 'entries', 'assign', 'freeze', 'seal', 'hasOwnProperty', 'toString', 'valueOf', + // Python 'print', 'len', 'range', 'str', 'int', 'float', 'list', 'dict', 'set', 'tuple', 'open', 'read', 'write', 'close', 'append', 'extend', 'update', 'super', 'type', 'isinstance', 'issubclass', 'getattr', 'setattr', 'hasattr', 'enumerate', 'zip', 'sorted', 'reversed', 'min', 'max', 'sum', 'abs', + // C/C++ standard library + 'printf', 'fprintf', 'sprintf', 'snprintf', 'vprintf', 'vfprintf', 'vsprintf', 'vsnprintf', + 'scanf', 'fscanf', 'sscanf', + 'malloc', 'calloc', 'realloc', 'free', 'memcpy', 'memmove', 'memset', 'memcmp', + 'strlen', 'strcpy', 'strncpy', 'strcat', 'strncat', 'strcmp', 'strncmp', 'strstr', 'strchr', 'strrchr', + 'atoi', 'atol', 'atof', 'strtol', 'strtoul', 'strtoll', 'strtoull', 'strtod', + 'sizeof', 'offsetof', 'typeof', + 'assert', 'abort', 'exit', '_exit', + 'fopen', 'fclose', 'fread', 'fwrite', 'fseek', 'ftell', 'rewind', 'fflush', 'fgets', 'fputs', + // Linux kernel common macros/helpers (not real call targets) + 'likely', 'unlikely', 'BUG', 'BUG_ON', 'WARN', 'WARN_ON', 'WARN_ONCE', + 'IS_ERR', 'PTR_ERR', 'ERR_PTR', 'IS_ERR_OR_NULL', + 'ARRAY_SIZE', 'container_of', 'list_for_each_entry', 'list_for_each_entry_safe', + 'min', 'max', 'clamp', 'abs', 'swap', + 'pr_info', 'pr_warn', 'pr_err', 'pr_debug', 'pr_notice', 'pr_crit', 'pr_emerg', + 'printk', 'dev_info', 'dev_warn', 'dev_err', 'dev_dbg', + 'GFP_KERNEL', 'GFP_ATOMIC', + 'spin_lock', 'spin_unlock', 'spin_lock_irqsave', 'spin_unlock_irqrestore', + 'mutex_lock', 'mutex_unlock', 'mutex_init', + 'kfree', 'kmalloc', 'kzalloc', 'kcalloc', 'krealloc', 'kvmalloc', 'kvfree', + 'get', 'put', ]); // ============================================================================ @@ -527,15 +550,57 @@ const processFileGroup = ( }; // ============================================================================ -// Worker message handler +// Worker message handler — supports sub-batch streaming // ============================================================================ -parentPort!.on('message', (files: ParseWorkerInput[]) => { +/** Accumulated result across sub-batches */ +let accumulated: ParseWorkerResult = { + nodes: [], relationships: [], symbols: [], + imports: [], calls: [], heritage: [], fileCount: 0, +}; +let cumulativeProcessed = 0; + +const mergeResult = (target: ParseWorkerResult, src: ParseWorkerResult) => { + target.nodes.push(...src.nodes); + target.relationships.push(...src.relationships); + target.symbols.push(...src.symbols); + target.imports.push(...src.imports); + target.calls.push(...src.calls); + target.heritage.push(...src.heritage); + target.fileCount += src.fileCount; +}; + +parentPort!.on('message', (msg: any) => { try { - const result = processBatch(files, (filesProcessed) => { - parentPort!.postMessage({ type: 'progress', filesProcessed }); - }); - parentPort!.postMessage({ type: 'result', data: result }); + // Sub-batch mode: { type: 'sub-batch', files: [...] } + if (msg && msg.type === 'sub-batch') { + const result = processBatch(msg.files, (filesProcessed) => { + parentPort!.postMessage({ type: 'progress', filesProcessed: cumulativeProcessed + filesProcessed }); + }); + cumulativeProcessed += result.fileCount; + mergeResult(accumulated, result); + // Signal ready for next sub-batch + parentPort!.postMessage({ type: 'sub-batch-done' }); + return; + } + + // Flush: send accumulated results + if (msg && msg.type === 'flush') { + parentPort!.postMessage({ type: 'result', data: accumulated }); + // Reset for potential reuse + accumulated = { nodes: [], relationships: [], symbols: [], imports: [], calls: [], heritage: [], fileCount: 0 }; + cumulativeProcessed = 0; + return; + } + + // Legacy single-message mode (backward compat): array of files + if (Array.isArray(msg)) { + const result = processBatch(msg, (filesProcessed) => { + parentPort!.postMessage({ type: 'progress', filesProcessed }); + }); + parentPort!.postMessage({ type: 'result', data: result }); + return; + } } catch (err) { const message = err instanceof Error ? err.message : String(err); parentPort!.postMessage({ type: 'error', error: message }); diff --git a/gitnexus/src/core/ingestion/workers/worker-pool.ts b/gitnexus/src/core/ingestion/workers/worker-pool.ts index 5c548dd75..1c1d7cae8 100644 --- a/gitnexus/src/core/ingestion/workers/worker-pool.ts +++ b/gitnexus/src/core/ingestion/workers/worker-pool.ts @@ -4,29 +4,33 @@ import os from 'node:os'; export interface WorkerPool { /** * Dispatch items across workers. Items are split into chunks (one per worker), - * each worker processes its chunk, and results are concatenated back in order. - * - * @param onProgress - Called with cumulative files processed across all workers + * each worker processes its chunk via sub-batches to limit peak memory, + * and results are concatenated back in order. */ dispatch(items: TInput[], onProgress?: (filesProcessed: number) => void): Promise; - /** - * Terminate all workers. Must be called when done. - */ + /** Terminate all workers. Must be called when done. */ terminate(): Promise; /** Number of workers in the pool */ readonly size: number; } +/** + * Max files to send to a worker in a single postMessage. + * Keeps structured-clone memory bounded per sub-batch. + */ +const SUB_BATCH_SIZE = 1500; + +/** Per sub-batch timeout. If a single sub-batch takes longer than this, + * likely a pathological file (e.g. minified 50MB JS). Fail fast. */ +const SUB_BATCH_TIMEOUT_MS = 30_000; + /** * Create a pool of worker threads. - * - * @param workerUrl - URL to the worker script (use `new URL('./parse-worker.js', import.meta.url)`) - * @param poolSize - Number of workers (defaults to cpus - 1, minimum 1) */ export const createWorkerPool = (workerUrl: URL, poolSize?: number): WorkerPool => { - const size = poolSize ?? Math.max(1, os.cpus().length - 1); + const size = poolSize ?? Math.min(8, Math.max(1, os.cpus().length - 1)); const workers: Worker[] = []; for (let i = 0; i < size; i++) { @@ -36,35 +40,51 @@ export const createWorkerPool = (workerUrl: URL, poolSize?: number): WorkerPool const dispatch = (items: TInput[], onProgress?: (filesProcessed: number) => void): Promise => { if (items.length === 0) return Promise.resolve([]); - // Split items into one chunk per worker const chunkSize = Math.ceil(items.length / size); const chunks: TInput[][] = []; for (let i = 0; i < items.length; i += chunkSize) { chunks.push(items.slice(i, i + chunkSize)); } - // Track per-worker progress for cumulative reporting const workerProgress = new Array(chunks.length).fill(0); - // Send one chunk to each worker, collect results const promises = chunks.map((chunk, i) => { const worker = workers[i]; return new Promise((resolve, reject) => { let settled = false; + let subBatchTimer: ReturnType | null = null; + const cleanup = () => { - clearTimeout(timer); + if (subBatchTimer) clearTimeout(subBatchTimer); worker.removeListener('message', handler); worker.removeListener('error', errorHandler); worker.removeListener('exit', exitHandler); }; - const timer = setTimeout(() => { - if (!settled) { - settled = true; - cleanup(); - reject(new Error(`Worker ${i} timed out after 5 minutes (chunk: ${chunk.length} items). Worker may have crashed or is processing too much data.`)); + const resetSubBatchTimer = () => { + if (subBatchTimer) clearTimeout(subBatchTimer); + subBatchTimer = setTimeout(() => { + if (!settled) { + settled = true; + cleanup(); + reject(new Error(`Worker ${i} sub-batch timed out after ${SUB_BATCH_TIMEOUT_MS / 1000}s (chunk: ${chunk.length} items).`)); + } + }, SUB_BATCH_TIMEOUT_MS); + }; + + let subBatchIdx = 0; + + const sendNextSubBatch = () => { + const start = subBatchIdx * SUB_BATCH_SIZE; + if (start >= chunk.length) { + worker.postMessage({ type: 'flush' }); + return; } - }, 5 * 60 * 1000); + const subBatch = chunk.slice(start, start + SUB_BATCH_SIZE); + subBatchIdx++; + resetSubBatchTimer(); + worker.postMessage({ type: 'sub-batch', files: subBatch }); + }; const handler = (msg: any) => { if (settled) return; @@ -74,8 +94,9 @@ export const createWorkerPool = (workerUrl: URL, poolSize?: number): WorkerPool const total = workerProgress.reduce((a, b) => a + b, 0); onProgress(total); } + } else if (msg && msg.type === 'sub-batch-done') { + sendNextSubBatch(); } else if (msg && msg.type === 'error') { - // Error reported by worker via postMessage settled = true; cleanup(); reject(new Error(`Worker ${i} error: ${msg.error}`)); @@ -84,7 +105,6 @@ export const createWorkerPool = (workerUrl: URL, poolSize?: number): WorkerPool cleanup(); resolve(msg.data); } else { - // Legacy: treat any non-typed message as result settled = true; cleanup(); resolve(msg); @@ -92,25 +112,21 @@ export const createWorkerPool = (workerUrl: URL, poolSize?: number): WorkerPool }; const errorHandler = (err: any) => { - if (!settled) { - settled = true; - cleanup(); - reject(err); - } + if (!settled) { settled = true; cleanup(); reject(err); } }; const exitHandler = (code: number) => { if (!settled) { settled = true; cleanup(); - reject(new Error(`Worker ${i} exited unexpectedly with code ${code}. This usually indicates an out-of-memory crash or native addon failure.`)); + reject(new Error(`Worker ${i} exited with code ${code}. Likely OOM or native addon failure.`)); } }; worker.on('message', handler); worker.once('error', errorHandler); worker.once('exit', exitHandler); - worker.postMessage(chunk); + sendNextSubBatch(); }); }); diff --git a/gitnexus/src/core/kuzu/csv-generator.ts b/gitnexus/src/core/kuzu/csv-generator.ts index 266309570..03175f59e 100644 --- a/gitnexus/src/core/kuzu/csv-generator.ts +++ b/gitnexus/src/core/kuzu/csv-generator.ts @@ -1,285 +1,340 @@ /** * CSV Generator for KuzuDB Hybrid Schema - * - * Generates separate CSV files for each node table and one relation CSV. - * This enables efficient bulk loading via COPY FROM for hybrid schema. - * + * + * Streams CSV rows directly to disk files in a single pass over graph nodes. + * File contents are lazy-read from disk per-node to avoid holding the entire + * repo in RAM. Rows are buffered (FLUSH_EVERY) before writing to minimize + * per-row Promise overhead. + * * RFC 4180 Compliant: * - Fields containing commas, double quotes, or newlines are enclosed in double quotes * - Double quotes within fields are escaped by doubling them ("") * - All fields are consistently quoted for safety with code content */ +import fs from 'fs/promises'; +import { createWriteStream, WriteStream } from 'fs'; +import path from 'path'; import { KnowledgeGraph, GraphNode, NodeLabel } from '../graph/types.js'; -import { NODE_TABLES, NodeTableName } from './schema.js'; +import { NodeTableName } from './schema.js'; + +/** Flush buffered rows to disk every N rows */ +const FLUSH_EVERY = 500; // ============================================================================ // CSV ESCAPE UTILITIES // ============================================================================ -/** - * Sanitize string to ensure valid UTF-8 and safe CSV content for KuzuDB - * Removes or replaces invalid characters that would break CSV parsing. - * - * Critical: KuzuDB's native CSV parser on Windows can misinterpret \r\n - * inside quoted fields. We normalize all line endings to \n only. - */ const sanitizeUTF8 = (str: string): string => { return str - .replace(/\r\n/g, '\n') // Normalize Windows line endings first - .replace(/\r/g, '\n') // Normalize remaining \r to \n - .replace(/[\x00-\x08\x0B\x0C\x0E-\x1F\x7F]/g, '') // Remove control chars except \t \n - .replace(/[\uD800-\uDFFF]/g, '') // Remove surrogate pairs (invalid standalone) - .replace(/[\uFFFE\uFFFF]/g, ''); // Remove BOM and special chars + .replace(/\r\n/g, '\n') + .replace(/\r/g, '\n') + .replace(/[\x00-\x08\x0B\x0C\x0E-\x1F\x7F]/g, '') + .replace(/[\uD800-\uDFFF]/g, '') + .replace(/[\uFFFE\uFFFF]/g, ''); }; -/** - * RFC 4180 compliant CSV field escaping - * ALWAYS wraps in double quotes for safety with code content - */ const escapeCSVField = (value: string | number | undefined | null): string => { - if (value === undefined || value === null) { - return '""'; - } + if (value === undefined || value === null) return '""'; let str = String(value); str = sanitizeUTF8(str); return `"${str.replace(/"/g, '""')}"`; }; -/** - * Escape a numeric value (no quotes needed for numbers) - */ const escapeCSVNumber = (value: number | undefined | null, defaultValue: number = -1): string => { - if (value === undefined || value === null) { - return String(defaultValue); - } + if (value === undefined || value === null) return String(defaultValue); return String(value); }; // ============================================================================ -// CONTENT EXTRACTION +// CONTENT EXTRACTION (lazy — reads from disk on demand) // ============================================================================ -/** - * Check if content looks like binary data - */ const isBinaryContent = (content: string): boolean => { if (!content || content.length === 0) return false; const sample = content.slice(0, 1000); let nonPrintable = 0; for (let i = 0; i < sample.length; i++) { const code = sample.charCodeAt(i); - if ((code < 9) || (code > 13 && code < 32) || code === 127) { - nonPrintable++; - } + if ((code < 9) || (code > 13 && code < 32) || code === 127) nonPrintable++; } return (nonPrintable / sample.length) > 0.1; }; /** - * Extract code content for a node + * LRU content cache — avoids re-reading the same source file for every + * symbol defined in it. Sized generously so most files stay cached during + * the single-pass node iteration. */ -const extractContent = ( +class FileContentCache { + private cache = new Map(); + private accessOrder: string[] = []; + private maxSize: number; + private repoPath: string; + + constructor(repoPath: string, maxSize: number = 3000) { + this.repoPath = repoPath; + this.maxSize = maxSize; + } + + async get(relativePath: string): Promise { + if (!relativePath) return ''; + const cached = this.cache.get(relativePath); + if (cached !== undefined) return cached; + try { + const fullPath = path.join(this.repoPath, relativePath); + const content = await fs.readFile(fullPath, 'utf-8'); + this.set(relativePath, content); + return content; + } catch { + this.set(relativePath, ''); + return ''; + } + } + + private set(key: string, value: string) { + if (this.cache.size >= this.maxSize) { + const oldest = this.accessOrder.shift(); + if (oldest) this.cache.delete(oldest); + } + this.cache.set(key, value); + this.accessOrder.push(key); + } +} + +const extractContent = async ( node: GraphNode, - fileContents: Map -): string => { + contentCache: FileContentCache +): Promise => { const filePath = node.properties.filePath; - const content = fileContents.get(filePath); - + const content = await contentCache.get(filePath); if (!content) return ''; if (node.label === 'Folder') return ''; if (isBinaryContent(content)) return '[Binary file - content not stored]'; - - // For File nodes, return content (limited) + if (node.label === 'File') { const MAX_FILE_CONTENT = 10000; - if (content.length > MAX_FILE_CONTENT) { - return content.slice(0, MAX_FILE_CONTENT) + '\n... [truncated]'; - } - return content; + return content.length > MAX_FILE_CONTENT + ? content.slice(0, MAX_FILE_CONTENT) + '\n... [truncated]' + : content; } - - // For code elements, extract the relevant lines with context + const startLine = node.properties.startLine; const endLine = node.properties.endLine; - if (startLine === undefined || endLine === undefined) return ''; - + const lines = content.split('\n'); - const contextLines = 2; - const start = Math.max(0, startLine - contextLines); - const end = Math.min(lines.length - 1, endLine + contextLines); - + const start = Math.max(0, startLine - 2); + const end = Math.min(lines.length - 1, endLine + 2); const snippet = lines.slice(start, end + 1).join('\n'); const MAX_SNIPPET = 5000; - if (snippet.length > MAX_SNIPPET) { - return snippet.slice(0, MAX_SNIPPET) + '\n... [truncated]'; - } - return snippet; + return snippet.length > MAX_SNIPPET + ? snippet.slice(0, MAX_SNIPPET) + '\n... [truncated]' + : snippet; }; // ============================================================================ -// CSV GENERATION RESULT TYPE +// BUFFERED CSV WRITER // ============================================================================ -export interface CSVData { - nodes: Map; - relCSV: string; // Single relation CSV with from,to,type,confidence,reason columns +class BufferedCSVWriter { + private ws: WriteStream; + private buffer: string[] = []; + rows = 0; + + constructor(filePath: string, header: string) { + this.ws = createWriteStream(filePath, 'utf-8'); + // Large repos flush many times — raise listener cap to avoid MaxListenersExceededWarning + this.ws.setMaxListeners(50); + this.buffer.push(header); + } + + addRow(row: string) { + this.buffer.push(row); + this.rows++; + if (this.buffer.length >= FLUSH_EVERY) { + return this.flush(); + } + return Promise.resolve(); + } + + flush(): Promise { + if (this.buffer.length === 0) return Promise.resolve(); + const chunk = this.buffer.join('\n') + '\n'; + this.buffer.length = 0; + return new Promise((resolve, reject) => { + const ok = this.ws.write(chunk); + if (ok) resolve(); + else this.ws.once('drain', resolve); + }); + } + + async finish(): Promise { + await this.flush(); + return new Promise((resolve, reject) => { + this.ws.end(() => resolve()); + this.ws.on('error', reject); + }); + } } // ============================================================================ -// NODE CSV GENERATORS +// STREAMING CSV GENERATION — SINGLE PASS // ============================================================================ -/** - * Generate CSV for File nodes - * Headers: id,name,filePath,content - */ -const generateFileCSV = (nodes: GraphNode[], fileContents: Map): string => { - const headers = ['id', 'name', 'filePath', 'content']; - const rows: string[] = [headers.join(',')]; - const seenIds = new Set(); - - for (const node of nodes) { - if (node.label !== 'File') continue; - // Skip duplicates - if (seenIds.has(node.id)) continue; - seenIds.add(node.id); - - const content = extractContent(node, fileContents); - rows.push([ - escapeCSVField(node.id), - escapeCSVField(node.properties.name || ''), - escapeCSVField(node.properties.filePath || ''), - escapeCSVField(content), - ].join(',')); - } - - return rows.join('\n'); -}; +export interface StreamedCSVResult { + nodeFiles: Map; + relCsvPath: string; + relRows: number; +} /** - * Generate CSV for Folder nodes - * Headers: id,name,filePath + * Stream all CSV data directly to disk files. + * Iterates graph nodes exactly ONCE — routes each node to the right writer. + * File contents are lazy-read from disk with a generous LRU cache. */ -const generateFolderCSV = (nodes: GraphNode[]): string => { - const headers = ['id', 'name', 'filePath']; - const rows: string[] = [headers.join(',')]; - - for (const node of nodes) { - if (node.label !== 'Folder') continue; - rows.push([ - escapeCSVField(node.id), - escapeCSVField(node.properties.name || ''), - escapeCSVField(node.properties.filePath || ''), - ].join(',')); - } - - return rows.join('\n'); -}; +export const streamAllCSVsToDisk = async ( + graph: KnowledgeGraph, + repoPath: string, + csvDir: string, +): Promise => { + // Remove stale CSVs from previous crashed runs, then recreate + try { await fs.rm(csvDir, { recursive: true, force: true }); } catch {} + await fs.mkdir(csvDir, { recursive: true }); -/** - * Generate CSV for code element nodes (Function, Class, Interface, Method, CodeElement) - * Headers: id,name,filePath,startLine,endLine,isExported,content - */ -const generateCodeElementCSV = ( - nodes: GraphNode[], - label: NodeLabel, - fileContents: Map -): string => { - const headers = ['id', 'name', 'filePath', 'startLine', 'endLine', 'isExported', 'content']; - const rows: string[] = [headers.join(',')]; - - for (const node of nodes) { - if (node.label !== label) continue; - const content = extractContent(node, fileContents); - rows.push([ - escapeCSVField(node.id), - escapeCSVField(node.properties.name || ''), - escapeCSVField(node.properties.filePath || ''), - escapeCSVNumber(node.properties.startLine, -1), - escapeCSVNumber(node.properties.endLine, -1), - node.properties.isExported ? 'true' : 'false', - escapeCSVField(content), - ].join(',')); - } - - return rows.join('\n'); -}; + // We open ~30 concurrent write-streams; raise process limit to suppress + // MaxListenersExceededWarning (restored after all streams finish). + const prevMax = process.getMaxListeners(); + process.setMaxListeners(prevMax + 40); -/** - * Generate CSV for Community nodes (from Leiden algorithm) - * Headers: id,label,heuristicLabel,keywords,description,enrichedBy,cohesion,symbolCount - */ -const generateCommunityCSV = (nodes: GraphNode[]): string => { - const headers = ['id', 'label', 'heuristicLabel', 'keywords', 'description', 'enrichedBy', 'cohesion', 'symbolCount']; - const rows: string[] = [headers.join(',')]; - - for (const node of nodes) { - if (node.label !== 'Community') continue; - - // Handle keywords array - convert to KuzuDB array format - const keywords = (node.properties as any).keywords || []; - const keywordsStr = `[${keywords.map((k: string) => `'${k.replace(/'/g, "''")}'`).join(',')}]`; - - rows.push([ - escapeCSVField(node.id), - escapeCSVField(node.properties.name || ''), // label is stored in name - escapeCSVField(node.properties.heuristicLabel || ''), - keywordsStr, // Array format for KuzuDB - escapeCSVField((node.properties as any).description || ''), - escapeCSVField((node.properties as any).enrichedBy || 'heuristic'), - escapeCSVNumber(node.properties.cohesion, 0), - escapeCSVNumber(node.properties.symbolCount, 0), - ].join(',')); - } - - return rows.join('\n'); -}; + const contentCache = new FileContentCache(repoPath); -/** - * Generate CSV for Process nodes - * Headers: id,label,heuristicLabel,processType,stepCount,communities,entryPointId,terminalId - */ -const generateProcessCSV = (nodes: GraphNode[]): string => { - const headers = ['id', 'label', 'heuristicLabel', 'processType', 'stepCount', 'communities', 'entryPointId', 'terminalId']; - const rows: string[] = [headers.join(',')]; - - for (const node of nodes) { - if (node.label !== 'Process') continue; - - // Handle communities array (string[]) - const communities = (node.properties as any).communities || []; - const communitiesStr = `[${communities.map((c: string) => `'${c.replace(/'/g, "''")}'`).join(',')}]`; - - rows.push([ - escapeCSVField(node.id), - escapeCSVField(node.properties.name || ''), // label stores name - escapeCSVField((node.properties as any).heuristicLabel || ''), - escapeCSVField((node.properties as any).processType || ''), - escapeCSVNumber((node.properties as any).stepCount, 0), - escapeCSVField(communitiesStr), // Needs CSV escaping because it contains commas! - escapeCSVField((node.properties as any).entryPointId || ''), - escapeCSVField((node.properties as any).terminalId || ''), - ].join(',')); - } - - return rows.join('\n'); -}; + // Create writers for every node type up-front + const fileWriter = new BufferedCSVWriter(path.join(csvDir, 'file.csv'), 'id,name,filePath,content'); + const folderWriter = new BufferedCSVWriter(path.join(csvDir, 'folder.csv'), 'id,name,filePath'); + const codeElementHeader = 'id,name,filePath,startLine,endLine,isExported,content'; + const functionWriter = new BufferedCSVWriter(path.join(csvDir, 'function.csv'), codeElementHeader); + const classWriter = new BufferedCSVWriter(path.join(csvDir, 'class.csv'), codeElementHeader); + const interfaceWriter = new BufferedCSVWriter(path.join(csvDir, 'interface.csv'), codeElementHeader); + const methodWriter = new BufferedCSVWriter(path.join(csvDir, 'method.csv'), codeElementHeader); + const codeElemWriter = new BufferedCSVWriter(path.join(csvDir, 'codeelement.csv'), codeElementHeader); + const communityWriter = new BufferedCSVWriter(path.join(csvDir, 'community.csv'), 'id,label,heuristicLabel,keywords,description,enrichedBy,cohesion,symbolCount'); + const processWriter = new BufferedCSVWriter(path.join(csvDir, 'process.csv'), 'id,label,heuristicLabel,processType,stepCount,communities,entryPointId,terminalId'); -/** - * Generate CSV for the single CodeRelation table - * Headers: from,to,type,confidence,reason - * - * confidence: 0-1 score for CALLS edges (how sure are we about the target?) - * reason: 'import-resolved' | 'same-file' | 'fuzzy-global' (or empty for non-CALLS) - */ -const generateRelationCSV = (graph: KnowledgeGraph): string => { - const headers = ['from', 'to', 'type', 'confidence', 'reason', 'step']; - const rows: string[] = [headers.join(',')]; - - for (const rel of graph.relationships) { - rows.push([ + // Multi-language node types share the same CSV shape (no isExported column) + const multiLangHeader = 'id,name,filePath,startLine,endLine,content'; + const MULTI_LANG_TYPES = ['Struct', 'Enum', 'Macro', 'Typedef', 'Union', 'Namespace', 'Trait', 'Impl', + 'TypeAlias', 'Const', 'Static', 'Property', 'Record', 'Delegate', 'Annotation', 'Constructor', 'Template', 'Module'] as const; + const multiLangWriters = new Map(); + for (const t of MULTI_LANG_TYPES) { + multiLangWriters.set(t, new BufferedCSVWriter(path.join(csvDir, `${t.toLowerCase()}.csv`), multiLangHeader)); + } + + const codeWriterMap: Record = { + 'Function': functionWriter, + 'Class': classWriter, + 'Interface': interfaceWriter, + 'Method': methodWriter, + 'CodeElement': codeElemWriter, + }; + + const seenFileIds = new Set(); + + // --- SINGLE PASS over all nodes --- + for (const node of graph.iterNodes()) { + switch (node.label) { + case 'File': { + if (seenFileIds.has(node.id)) break; + seenFileIds.add(node.id); + const content = await extractContent(node, contentCache); + await fileWriter.addRow([ + escapeCSVField(node.id), + escapeCSVField(node.properties.name || ''), + escapeCSVField(node.properties.filePath || ''), + escapeCSVField(content), + ].join(',')); + break; + } + case 'Folder': + await folderWriter.addRow([ + escapeCSVField(node.id), + escapeCSVField(node.properties.name || ''), + escapeCSVField(node.properties.filePath || ''), + ].join(',')); + break; + case 'Community': { + const keywords = (node.properties as any).keywords || []; + const keywordsStr = `[${keywords.map((k: string) => `'${k.replace(/'/g, "''")}'`).join(',')}]`; + await communityWriter.addRow([ + escapeCSVField(node.id), + escapeCSVField(node.properties.name || ''), + escapeCSVField(node.properties.heuristicLabel || ''), + keywordsStr, + escapeCSVField((node.properties as any).description || ''), + escapeCSVField((node.properties as any).enrichedBy || 'heuristic'), + escapeCSVNumber(node.properties.cohesion, 0), + escapeCSVNumber(node.properties.symbolCount, 0), + ].join(',')); + break; + } + case 'Process': { + const communities = (node.properties as any).communities || []; + const communitiesStr = `[${communities.map((c: string) => `'${c.replace(/'/g, "''")}'`).join(',')}]`; + await processWriter.addRow([ + escapeCSVField(node.id), + escapeCSVField(node.properties.name || ''), + escapeCSVField((node.properties as any).heuristicLabel || ''), + escapeCSVField((node.properties as any).processType || ''), + escapeCSVNumber((node.properties as any).stepCount, 0), + escapeCSVField(communitiesStr), + escapeCSVField((node.properties as any).entryPointId || ''), + escapeCSVField((node.properties as any).terminalId || ''), + ].join(',')); + break; + } + default: { + // Code element nodes (Function, Class, Interface, Method, CodeElement) + const writer = codeWriterMap[node.label]; + if (writer) { + const content = await extractContent(node, contentCache); + await writer.addRow([ + escapeCSVField(node.id), + escapeCSVField(node.properties.name || ''), + escapeCSVField(node.properties.filePath || ''), + escapeCSVNumber(node.properties.startLine, -1), + escapeCSVNumber(node.properties.endLine, -1), + node.properties.isExported ? 'true' : 'false', + escapeCSVField(content), + ].join(',')); + } else { + // Multi-language node types (Struct, Impl, Trait, Macro, etc.) + const mlWriter = multiLangWriters.get(node.label); + if (mlWriter) { + const content = await extractContent(node, contentCache); + await mlWriter.addRow([ + escapeCSVField(node.id), + escapeCSVField(node.properties.name || ''), + escapeCSVField(node.properties.filePath || ''), + escapeCSVNumber(node.properties.startLine, -1), + escapeCSVNumber(node.properties.endLine, -1), + escapeCSVField(content), + ].join(',')); + } + } + break; + } + } + } + + // Finish all node writers + const allWriters = [fileWriter, folderWriter, functionWriter, classWriter, interfaceWriter, methodWriter, codeElemWriter, communityWriter, processWriter, ...multiLangWriters.values()]; + await Promise.all(allWriters.map(w => w.finish())); + + // --- Stream relationship CSV --- + const relCsvPath = path.join(csvDir, 'relations.csv'); + const relWriter = new BufferedCSVWriter(relCsvPath, 'from,to,type,confidence,reason,step'); + for (const rel of graph.iterRelationships()) { + await relWriter.addRow([ escapeCSVField(rel.sourceId), escapeCSVField(rel.targetId), escapeCSVField(rel.type), @@ -288,39 +343,26 @@ const generateRelationCSV = (graph: KnowledgeGraph): string => { escapeCSVNumber((rel as any).step, 0), ].join(',')); } - - return rows.join('\n'); + await relWriter.finish(); + + // Build result map — only include tables that have rows + const nodeFiles = new Map(); + const tableMap: [NodeTableName, BufferedCSVWriter][] = [ + ['File', fileWriter], ['Folder', folderWriter], + ['Function', functionWriter], ['Class', classWriter], + ['Interface', interfaceWriter], ['Method', methodWriter], + ['CodeElement', codeElemWriter], + ['Community', communityWriter], ['Process', processWriter], + ...Array.from(multiLangWriters.entries()).map(([name, w]) => [name as NodeTableName, w] as [NodeTableName, BufferedCSVWriter]), + ]; + for (const [name, writer] of tableMap) { + if (writer.rows > 0) { + nodeFiles.set(name, { csvPath: path.join(csvDir, `${name.toLowerCase()}.csv`), rows: writer.rows }); + } + } + + // Restore original process listener limit + process.setMaxListeners(prevMax); + + return { nodeFiles, relCsvPath, relRows: relWriter.rows }; }; - -// ============================================================================ -// MAIN CSV GENERATION FUNCTION -// ============================================================================ - -/** - * Generate all CSV data for hybrid schema bulk loading - * Returns Maps of node table name -> CSV content, and single relation CSV - */ -export const generateAllCSVs = ( - graph: KnowledgeGraph, - fileContents: Map -): CSVData => { - const nodes = Array.from(graph.nodes); - - // Generate node CSVs - const nodeCSVs = new Map(); - nodeCSVs.set('File', generateFileCSV(nodes, fileContents)); - nodeCSVs.set('Folder', generateFolderCSV(nodes)); - nodeCSVs.set('Function', generateCodeElementCSV(nodes, 'Function', fileContents)); - nodeCSVs.set('Class', generateCodeElementCSV(nodes, 'Class', fileContents)); - nodeCSVs.set('Interface', generateCodeElementCSV(nodes, 'Interface', fileContents)); - nodeCSVs.set('Method', generateCodeElementCSV(nodes, 'Method', fileContents)); - nodeCSVs.set('CodeElement', generateCodeElementCSV(nodes, 'CodeElement', fileContents)); - nodeCSVs.set('Community', generateCommunityCSV(nodes)); - nodeCSVs.set('Process', generateProcessCSV(nodes)); - - // Generate single relation CSV - const relCSV = generateRelationCSV(graph); - - return { nodes: nodeCSVs, relCSV }; -}; - diff --git a/gitnexus/src/core/kuzu/kuzu-adapter.ts b/gitnexus/src/core/kuzu/kuzu-adapter.ts index 9988b56ed..797144793 100644 --- a/gitnexus/src/core/kuzu/kuzu-adapter.ts +++ b/gitnexus/src/core/kuzu/kuzu-adapter.ts @@ -1,4 +1,6 @@ import fs from 'fs/promises'; +import { createReadStream } from 'fs'; +import { createInterface } from 'readline'; import path from 'path'; import kuzu from 'kuzu'; import { KnowledgeGraph } from '../graph/types.js'; @@ -9,7 +11,7 @@ import { EMBEDDING_TABLE_NAME, NodeTableName, } from './schema.js'; -import { generateAllCSVs } from './csv-generator.js'; +import { streamAllCSVsToDisk } from './csv-generator.js'; let db: kuzu.Database | null = null; let conn: kuzu.Connection | null = null; @@ -65,7 +67,7 @@ export type KuzuProgressCallback = (message: string) => void; export const loadGraphToKuzu = async ( graph: KnowledgeGraph, - fileContents: Map, + repoPath: string, storagePath: string, onProgress?: KuzuProgressCallback ) => { @@ -75,23 +77,11 @@ export const loadGraphToKuzu = async ( const log = onProgress || (() => {}); - const csvData = generateAllCSVs(graph, fileContents); const csvDir = path.join(storagePath, 'csv'); - await fs.mkdir(csvDir, { recursive: true }); - log('Generating CSVs...'); + log('Streaming CSVs to disk...'); + const csvResult = await streamAllCSVsToDisk(graph, repoPath, csvDir); - const nodeFiles: Array<{ table: NodeTableName; path: string; rows: number }> = []; - for (const [tableName, csv] of csvData.nodes.entries()) { - const rowCount = csv.split('\n').length - 1; - if (rowCount <= 0) continue; - const filePath = path.join(csvDir, `${tableName.toLowerCase()}.csv`); - await fs.writeFile(filePath, csv, 'utf-8'); - nodeFiles.push({ table: tableName, path: filePath, rows: rowCount }); - } - - // Write relationship CSV to disk for bulk COPY - const relCsvPath = path.join(csvDir, 'relations.csv'); const validTables = new Set(NODE_TABLES as readonly string[]); const getNodeLabel = (nodeId: string): string => { if (nodeId.startsWith('comm_')) return 'Community'; @@ -99,34 +89,16 @@ export const loadGraphToKuzu = async ( return nodeId.split(':')[0]; }; - const relLines = csvData.relCSV.split('\n'); - const relHeader = relLines[0]; - const validRelLines = [relHeader]; - let skippedRels = 0; - for (let i = 1; i < relLines.length; i++) { - const line = relLines[i]; - if (!line.trim()) continue; - const match = line.match(/"([^"]*)","([^"]*)"/); - if (!match) { skippedRels++; continue; } - const fromLabel = getNodeLabel(match[1]); - const toLabel = getNodeLabel(match[2]); - if (!validTables.has(fromLabel) || !validTables.has(toLabel)) { - skippedRels++; - continue; - } - validRelLines.push(line); - } - await fs.writeFile(relCsvPath, validRelLines.join('\n'), 'utf-8'); - - // Bulk COPY all node CSVs + // Bulk COPY all node CSVs (sequential — KuzuDB allows only one write txn at a time) + const nodeFiles = [...csvResult.nodeFiles.entries()]; const totalSteps = nodeFiles.length + 1; // +1 for relationships let stepsDone = 0; - for (const { table, path: filePath, rows } of nodeFiles) { + for (const [table, { csvPath, rows }] of nodeFiles) { stepsDone++; log(`Loading nodes ${stepsDone}/${totalSteps}: ${table} (${rows.toLocaleString()} rows)`); - const normalizedPath = normalizeCopyPath(filePath); + const normalizedPath = normalizeCopyPath(csvPath); const copyQuery = getCopyQuery(table, normalizedPath); try { @@ -143,21 +115,39 @@ export const loadGraphToKuzu = async ( } // Bulk COPY relationships — split by FROM→TO label pair (KuzuDB requires it) - const insertedRels = validRelLines.length - 1; - const warnings: string[] = []; - if (insertedRels > 0) { - const relsByPair = new Map(); - for (let i = 1; i < validRelLines.length; i++) { - const line = validRelLines[i]; + // Stream-read the relation CSV line by line to avoid exceeding V8 max string length + let relHeader = ''; + const relsByPair = new Map(); + let skippedRels = 0; + let totalValidRels = 0; + + await new Promise((resolve, reject) => { + const rl = createInterface({ input: createReadStream(csvResult.relCsvPath, 'utf-8'), crlfDelay: Infinity }); + let isFirst = true; + rl.on('line', (line) => { + if (isFirst) { relHeader = line; isFirst = false; return; } + if (!line.trim()) return; const match = line.match(/"([^"]*)","([^"]*)"/); - if (!match) continue; + if (!match) { skippedRels++; return; } const fromLabel = getNodeLabel(match[1]); const toLabel = getNodeLabel(match[2]); + if (!validTables.has(fromLabel) || !validTables.has(toLabel)) { + skippedRels++; + return; + } const pairKey = `${fromLabel}|${toLabel}`; let list = relsByPair.get(pairKey); if (!list) { list = []; relsByPair.set(pairKey, list); } list.push(line); - } + totalValidRels++; + }); + rl.on('close', resolve); + rl.on('error', reject); + }); + + const insertedRels = totalValidRels; + const warnings: string[] = []; + if (insertedRels > 0) { log(`Loading edges: ${insertedRels.toLocaleString()} across ${relsByPair.size} types`); @@ -200,9 +190,9 @@ export const loadGraphToKuzu = async ( } // Cleanup all CSVs - try { await fs.unlink(relCsvPath); } catch {} - for (const { path: filePath } of nodeFiles) { - try { await fs.unlink(filePath); } catch {} + try { await fs.unlink(csvResult.relCsvPath); } catch {} + for (const [, { csvPath }] of csvResult.nodeFiles) { + try { await fs.unlink(csvPath); } catch {} } try { const remaining = await fs.readdir(csvDir); @@ -268,6 +258,9 @@ const fallbackRelationshipInserts = async ( } }; +/** Tables with isExported column (TypeScript/JS-native types) */ +const TABLES_WITH_EXPORTED = new Set(['Function', 'Class', 'Interface', 'Method', 'CodeElement']); + const getCopyQuery = (table: NodeTableName, filePath: string): string => { const t = escapeTableName(table); if (table === 'File') { @@ -282,8 +275,12 @@ const getCopyQuery = (table: NodeTableName, filePath: string): string => { if (table === 'Process') { return `COPY ${t}(id, label, heuristicLabel, processType, stepCount, communities, entryPointId, terminalId) FROM "${filePath}" ${COPY_CSV_OPTS}`; } - // Code element tables (Function, Class, Interface, Method, CodeElement, and multi-language) - return `COPY ${t}(id, name, filePath, startLine, endLine, isExported, content) FROM "${filePath}" ${COPY_CSV_OPTS}`; + // TypeScript/JS code element tables have isExported; multi-language tables do not + if (TABLES_WITH_EXPORTED.has(table)) { + return `COPY ${t}(id, name, filePath, startLine, endLine, isExported, content) FROM "${filePath}" ${COPY_CSV_OPTS}`; + } + // Multi-language tables (Struct, Impl, Trait, Macro, etc.) + return `COPY ${t}(id, name, filePath, startLine, endLine, content) FROM "${filePath}" ${COPY_CSV_OPTS}`; }; /** @@ -312,15 +309,18 @@ export const insertNodeToKuzu = async ( }; // Build INSERT query based on node type + const t = escapeTableName(label); let query: string; - + if (label === 'File') { query = `CREATE (n:File {id: ${escapeValue(properties.id)}, name: ${escapeValue(properties.name)}, filePath: ${escapeValue(properties.filePath)}, content: ${escapeValue(properties.content || '')}})`; } else if (label === 'Folder') { query = `CREATE (n:Folder {id: ${escapeValue(properties.id)}, name: ${escapeValue(properties.name)}, filePath: ${escapeValue(properties.filePath)}})`; + } else if (TABLES_WITH_EXPORTED.has(label)) { + query = `CREATE (n:${t} {id: ${escapeValue(properties.id)}, name: ${escapeValue(properties.name)}, filePath: ${escapeValue(properties.filePath)}, startLine: ${properties.startLine || 0}, endLine: ${properties.endLine || 0}, isExported: ${!!properties.isExported}, content: ${escapeValue(properties.content || '')}})`; } else { - // Function, Class, Method, Interface, etc. - standard code element schema - query = `CREATE (n:${label} {id: ${escapeValue(properties.id)}, name: ${escapeValue(properties.name)}, filePath: ${escapeValue(properties.filePath)}, startLine: ${properties.startLine || 0}, endLine: ${properties.endLine || 0}, content: ${escapeValue(properties.content || '')}})`; + // Multi-language tables (Struct, Impl, Trait, Macro, etc.) — no isExported + query = `CREATE (n:${t} {id: ${escapeValue(properties.id)}, name: ${escapeValue(properties.name)}, filePath: ${escapeValue(properties.filePath)}, startLine: ${properties.startLine || 0}, endLine: ${properties.endLine || 0}, content: ${escapeValue(properties.content || '')}})`; } // Use per-query connection if dbPath provided (avoids lock conflicts) @@ -380,12 +380,15 @@ export const batchInsertNodesToKuzu = async ( let query: string; // Use MERGE instead of CREATE for upsert behavior (handles duplicates gracefully) + const t = escapeTableName(label); if (label === 'File') { query = `MERGE (n:File {id: ${escapeValue(properties.id)}}) SET n.name = ${escapeValue(properties.name)}, n.filePath = ${escapeValue(properties.filePath)}, n.content = ${escapeValue(properties.content || '')}`; } else if (label === 'Folder') { query = `MERGE (n:Folder {id: ${escapeValue(properties.id)}}) SET n.name = ${escapeValue(properties.name)}, n.filePath = ${escapeValue(properties.filePath)}`; + } else if (TABLES_WITH_EXPORTED.has(label)) { + query = `MERGE (n:${t} {id: ${escapeValue(properties.id)}}) SET n.name = ${escapeValue(properties.name)}, n.filePath = ${escapeValue(properties.filePath)}, n.startLine = ${properties.startLine || 0}, n.endLine = ${properties.endLine || 0}, n.isExported = ${!!properties.isExported}, n.content = ${escapeValue(properties.content || '')}`; } else { - query = `MERGE (n:${label} {id: ${escapeValue(properties.id)}}) SET n.name = ${escapeValue(properties.name)}, n.filePath = ${escapeValue(properties.filePath)}, n.startLine = ${properties.startLine || 0}, n.endLine = ${properties.endLine || 0}, n.content = ${escapeValue(properties.content || '')}`; + query = `MERGE (n:${t} {id: ${escapeValue(properties.id)}}) SET n.name = ${escapeValue(properties.name)}, n.filePath = ${escapeValue(properties.filePath)}, n.startLine = ${properties.startLine || 0}, n.endLine = ${properties.endLine || 0}, n.content = ${escapeValue(properties.content || '')}`; } await tempConn.query(query); @@ -451,7 +454,7 @@ export const getKuzuStats = async (): Promise<{ nodes: number; edges: number }> let totalNodes = 0; for (const tableName of NODE_TABLES) { try { - const queryResult = await conn.query(`MATCH (n:${tableName}) RETURN count(n) AS cnt`); + const queryResult = await conn.query(`MATCH (n:${escapeTableName(tableName)}) RETURN count(n) AS cnt`); const nodeResult = Array.isArray(queryResult) ? queryResult[0] : queryResult; const nodeRows = await nodeResult.getAll(); if (nodeRows.length > 0) { @@ -525,6 +528,7 @@ export const closeKuzu = async (): Promise => { } catch {} db = null; } + ftsLoaded = false; }; export const isKuzuReady = (): boolean => conn !== null && db !== null; @@ -563,17 +567,18 @@ export const deleteNodesForFile = async (filePath: string, dbPath?: string): Pro try { // First count how many we'll delete + const tn = escapeTableName(tableName); const countResult = await targetConn!.query( - `MATCH (n:${tableName}) WHERE n.filePath = '${escapedPath}' RETURN count(n) AS cnt` + `MATCH (n:${tn}) WHERE n.filePath = '${escapedPath}' RETURN count(n) AS cnt` ); const result = Array.isArray(countResult) ? countResult[0] : countResult; const rows = await result.getAll(); const count = Number(rows[0]?.cnt ?? rows[0]?.[0] ?? 0); - + if (count > 0) { // Delete nodes (and implicitly their relationships via DETACH) await targetConn!.query( - `MATCH (n:${tableName}) WHERE n.filePath = '${escapedPath}' DETACH DELETE n` + `MATCH (n:${tn}) WHERE n.filePath = '${escapedPath}' DETACH DELETE n` ); deletedNodes += count; } @@ -610,9 +615,12 @@ export const getEmbeddingTableName = (): string => EMBEDDING_TABLE_NAME; // ============================================================================ /** - * Load the FTS extension (required before using FTS functions) + * Load the FTS extension (required before using FTS functions). + * Safe to call multiple times — tracks loaded state. */ +let ftsLoaded = false; export const loadFTSExtension = async (): Promise => { + if (ftsLoaded) return; if (!conn) { throw new Error('KuzuDB not initialized. Call initKuzu first.'); } @@ -622,6 +630,7 @@ export const loadFTSExtension = async (): Promise => { } catch { // Extension may already be loaded } + ftsLoaded = true; }; /** @@ -640,16 +649,15 @@ export const createFTSIndex = async ( if (!conn) { throw new Error('KuzuDB not initialized. Call initKuzu first.'); } - + await loadFTSExtension(); - + const propList = properties.map(p => `'${p}'`).join(', '); const query = `CALL CREATE_FTS_INDEX('${tableName}', '${indexName}', [${propList}], stemmer := '${stemmer}')`; - + try { await conn.query(query); } catch (e: any) { - // Index may already exist if (!e.message?.includes('already exists')) { throw e; } diff --git a/gitnexus/src/core/kuzu/schema.ts b/gitnexus/src/core/kuzu/schema.ts index c0f3b394a..50cc45988 100644 --- a/gitnexus/src/core/kuzu/schema.ts +++ b/gitnexus/src/core/kuzu/schema.ts @@ -232,6 +232,9 @@ CREATE REL TABLE ${REL_TABLE_NAME} ( FROM Function TO \`Impl\`, FROM Function TO Interface, FROM Function TO \`Constructor\`, + FROM Function TO \`Const\`, + FROM Function TO \`Typedef\`, + FROM Function TO \`Union\`, FROM Class TO Method, FROM Class TO Function, FROM Class TO Class, @@ -243,6 +246,10 @@ CREATE REL TABLE ${REL_TABLE_NAME} ( FROM Class TO \`Enum\`, FROM Class TO \`Annotation\`, FROM Class TO \`Constructor\`, + FROM Class TO \`Trait\`, + FROM Class TO \`Macro\`, + FROM Class TO \`Impl\`, + FROM Class TO \`Union\`, FROM Method TO Function, FROM Method TO Method, FROM Method TO Class, @@ -293,7 +300,10 @@ CREATE REL TABLE ${REL_TABLE_NAME} ( FROM \`Trait\` TO Community, FROM \`Impl\` TO Community, FROM \`Impl\` TO \`Trait\`, + FROM \`Impl\` TO \`Struct\`, + FROM \`Impl\` TO \`Impl\`, FROM \`TypeAlias\` TO Community, + FROM \`TypeAlias\` TO \`Trait\`, FROM \`Const\` TO Community, FROM \`Static\` TO Community, FROM \`Property\` TO Community, diff --git a/gitnexus/src/types/pipeline.ts b/gitnexus/src/types/pipeline.ts index c8848d562..e8471c767 100644 --- a/gitnexus/src/types/pipeline.ts +++ b/gitnexus/src/types/pipeline.ts @@ -19,7 +19,10 @@ export interface PipelineProgress { // Original result type (used internally in pipeline) export interface PipelineResult { graph: KnowledgeGraph; - fileContents: Map; + /** Absolute path to the repo root — used for lazy file reads during KuzuDB loading */ + repoPath: string; + /** Total files scanned (for stats) */ + totalFileCount: number; communityResult?: CommunityDetectionResult; processResult?: ProcessDetectionResult; } @@ -29,14 +32,16 @@ export interface PipelineResult { export interface SerializablePipelineResult { nodes: GraphNode[]; relationships: GraphRelationship[]; - fileContents: Record; // Object instead of Map + repoPath: string; + totalFileCount: number; } // Helper to convert PipelineResult to serializable format export const serializePipelineResult = (result: PipelineResult): SerializablePipelineResult => ({ - nodes: result.graph.nodes, - relationships: result.graph.relationships, - fileContents: Object.fromEntries(result.fileContents), + nodes: [...result.graph.iterNodes()], + relationships: [...result.graph.iterRelationships()], + repoPath: result.repoPath, + totalFileCount: result.totalFileCount, }); // Helper to reconstruct from serializable format (used in main thread) @@ -47,10 +52,11 @@ export const deserializePipelineResult = ( const graph = createGraph(); serialized.nodes.forEach(node => graph.addNode(node)); serialized.relationships.forEach(rel => graph.addRelationship(rel)); - + return { graph, - fileContents: new Map(Object.entries(serialized.fileContents)), + repoPath: serialized.repoPath, + totalFileCount: serialized.totalFileCount, }; }; From c7e1c6536cce3f037e73212adf1a6ce73bbc677b Mon Sep 17 00:00:00 2001 From: abhigyanpatwari Date: Thu, 26 Feb 2026 10:28:38 +0530 Subject: [PATCH 4/7] chore: bump version to 1.3.0 Co-Authored-By: Claude Opus 4.6 --- gitnexus/package.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/gitnexus/package.json b/gitnexus/package.json index 261be4bfc..af5505cb5 100644 --- a/gitnexus/package.json +++ b/gitnexus/package.json @@ -1,6 +1,6 @@ { "name": "gitnexus", - "version": "1.2.9", + "version": "1.3.0", "description": "Graph-powered code intelligence for AI agents. Index any codebase, query via MCP or CLI.", "author": "Abhigyan Patwari", "license": "PolyForm-Noncommercial-1.0.0", From ee090204a032fab7b9997eab103cb4b7553e6c64 Mon Sep 17 00:00:00 2001 From: abhigyanpatwari Date: Thu, 26 Feb 2026 10:43:55 +0530 Subject: [PATCH 5/7] fix(schema): add 6 missing FROM/TO pairs for LLVM-style codebases MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Class→Namespace, Class→Typedef, Struct→Struct, Struct→Class, Struct→Enum, Namespace→Struct Co-Authored-By: Claude Opus 4.6 --- gitnexus/package.json | 2 +- gitnexus/src/core/kuzu/schema.ts | 6 ++++++ 2 files changed, 7 insertions(+), 1 deletion(-) diff --git a/gitnexus/package.json b/gitnexus/package.json index af5505cb5..891efe4a1 100644 --- a/gitnexus/package.json +++ b/gitnexus/package.json @@ -1,6 +1,6 @@ { "name": "gitnexus", - "version": "1.3.0", + "version": "1.3.1", "description": "Graph-powered code intelligence for AI agents. Index any codebase, query via MCP or CLI.", "author": "Abhigyan Patwari", "license": "PolyForm-Noncommercial-1.0.0", diff --git a/gitnexus/src/core/kuzu/schema.ts b/gitnexus/src/core/kuzu/schema.ts index 50cc45988..7fcdac826 100644 --- a/gitnexus/src/core/kuzu/schema.ts +++ b/gitnexus/src/core/kuzu/schema.ts @@ -250,6 +250,8 @@ CREATE REL TABLE ${REL_TABLE_NAME} ( FROM Class TO \`Macro\`, FROM Class TO \`Impl\`, FROM Class TO \`Union\`, + FROM Class TO \`Namespace\`, + FROM Class TO \`Typedef\`, FROM Method TO Function, FROM Method TO Method, FROM Method TO Class, @@ -286,6 +288,9 @@ CREATE REL TABLE ${REL_TABLE_NAME} ( FROM Interface TO \`Constructor\`, FROM \`Struct\` TO Community, FROM \`Struct\` TO \`Trait\`, + FROM \`Struct\` TO \`Struct\`, + FROM \`Struct\` TO Class, + FROM \`Struct\` TO \`Enum\`, FROM \`Struct\` TO Function, FROM \`Struct\` TO Method, FROM \`Enum\` TO Community, @@ -297,6 +302,7 @@ CREATE REL TABLE ${REL_TABLE_NAME} ( FROM \`Typedef\` TO Community, FROM \`Union\` TO Community, FROM \`Namespace\` TO Community, + FROM \`Namespace\` TO \`Struct\`, FROM \`Trait\` TO Community, FROM \`Impl\` TO Community, FROM \`Impl\` TO \`Trait\`, From 8ab3970150dd6ba8de4ca31ac9b6ecd41da8b0ff Mon Sep 17 00:00:00 2001 From: abhigyanpatwari Date: Thu, 26 Feb 2026 10:46:10 +0530 Subject: [PATCH 6/7] fix(ux): raise file size limit to 2MB, quiet warning output - Raise MAX_FILE_SIZE from 512KB to 2MB to capture more real source files - Replace verbose per-warning output with single summary line - Soften skip message wording ("likely generated/vendored") Co-Authored-By: Claude Opus 4.6 --- gitnexus/package.json | 2 +- gitnexus/src/cli/analyze.ts | 11 ++++++----- gitnexus/src/core/ingestion/filesystem-walker.ts | 6 +++--- gitnexus/src/core/ingestion/parsing-processor.ts | 2 +- gitnexus/src/core/ingestion/workers/parse-worker.ts | 2 +- 5 files changed, 12 insertions(+), 11 deletions(-) diff --git a/gitnexus/package.json b/gitnexus/package.json index 891efe4a1..26fdfce39 100644 --- a/gitnexus/package.json +++ b/gitnexus/package.json @@ -1,6 +1,6 @@ { "name": "gitnexus", - "version": "1.3.1", + "version": "1.3.2", "description": "Graph-powered code intelligence for AI agents. Index any codebase, query via MCP or CLI.", "author": "Abhigyan Patwari", "license": "PolyForm-Noncommercial-1.0.0", diff --git a/gitnexus/src/cli/analyze.ts b/gitnexus/src/cli/analyze.ts index d3fbf6ba7..6ed6c05ef 100644 --- a/gitnexus/src/cli/analyze.ts +++ b/gitnexus/src/cli/analyze.ts @@ -343,12 +343,13 @@ export const analyzeCommand = async ( console.log(` Hooks: ${hookResult.message}`); } - // Show warnings (missing schema pairs, etc.) after the clean output + // Show a quiet summary if some edge types needed fallback insertion if (kuzuWarnings.length > 0) { - console.log(`\n Warnings (${kuzuWarnings.length}):`); - for (const w of kuzuWarnings) { - console.log(` ${w}`); - } + const totalFallback = kuzuWarnings.reduce((sum, w) => { + const m = w.match(/\((\d+) edges\)/); + return sum + (m ? parseInt(m[1]) : 0); + }, 0); + console.log(` Note: ${totalFallback} edges across ${kuzuWarnings.length} types inserted via fallback (schema will be updated in next release)`); } try { diff --git a/gitnexus/src/core/ingestion/filesystem-walker.ts b/gitnexus/src/core/ingestion/filesystem-walker.ts index b17403238..c9365d3b0 100644 --- a/gitnexus/src/core/ingestion/filesystem-walker.ts +++ b/gitnexus/src/core/ingestion/filesystem-walker.ts @@ -21,8 +21,8 @@ export interface FilePath { const READ_CONCURRENCY = 32; -/** Skip files larger than 512KB — they're usually generated/vendored and crash tree-sitter */ -const MAX_FILE_SIZE = 512 * 1024; +/** Skip files larger than 2MB — covers all real source files, excludes minified bundles/generated code */ +const MAX_FILE_SIZE = 2 * 1024 * 1024; /** * Phase 1: Scan repository — stat files to get paths + sizes, no content loaded. @@ -69,7 +69,7 @@ export const walkRepositoryPaths = async ( } if (skippedLarge > 0) { - console.warn(` Skipped ${skippedLarge} files larger than ${MAX_FILE_SIZE / 1024}KB`); + console.warn(` Skipped ${skippedLarge} files larger than ${Math.round(MAX_FILE_SIZE / 1024 / 1024)}MB (likely generated/vendored)`); } return entries; diff --git a/gitnexus/src/core/ingestion/parsing-processor.ts b/gitnexus/src/core/ingestion/parsing-processor.ts index c15d39a17..1134ebe49 100644 --- a/gitnexus/src/core/ingestion/parsing-processor.ts +++ b/gitnexus/src/core/ingestion/parsing-processor.ts @@ -207,7 +207,7 @@ const processParsingSequential = async ( if (!language) continue; // Skip very large files — they can crash tree-sitter or cause OOM - if (file.content.length > 512 * 1024) continue; + if (file.content.length > 2 * 1024 * 1024) continue; await loadLanguage(language, file.path); diff --git a/gitnexus/src/core/ingestion/workers/parse-worker.ts b/gitnexus/src/core/ingestion/workers/parse-worker.ts index 2eee2c5bc..a89d8b402 100644 --- a/gitnexus/src/core/ingestion/workers/parse-worker.ts +++ b/gitnexus/src/core/ingestion/workers/parse-worker.ts @@ -426,7 +426,7 @@ const processFileGroup = ( for (const file of files) { // Skip very large files — they can crash tree-sitter or cause OOM - if (file.content.length > 512 * 1024) continue; + if (file.content.length > 2 * 1024 * 1024) continue; let tree; try { From d3556316e1cdab384f2459b6e98796d694d12c01 Mon Sep 17 00:00:00 2001 From: abhigyanpatwari Date: Thu, 26 Feb 2026 11:20:15 +0530 Subject: [PATCH 7/7] revert: restore 512KB file size limit, keep improved skip message 2MB limit caused FTS crash on large codebases. 512KB is safe and only skips generated/vendored files. Co-Authored-By: Claude Opus 4.6 --- gitnexus/package-lock.json | 4 ++-- gitnexus/package.json | 2 +- gitnexus/src/core/ingestion/filesystem-walker.ts | 6 +++--- gitnexus/src/core/ingestion/parsing-processor.ts | 2 +- gitnexus/src/core/ingestion/workers/parse-worker.ts | 2 +- 5 files changed, 8 insertions(+), 8 deletions(-) diff --git a/gitnexus/package-lock.json b/gitnexus/package-lock.json index b937d11b9..4d555d47b 100644 --- a/gitnexus/package-lock.json +++ b/gitnexus/package-lock.json @@ -1,12 +1,12 @@ { "name": "gitnexus", - "version": "1.2.9", + "version": "1.3.3", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "gitnexus", - "version": "1.2.9", + "version": "1.3.3", "license": "PolyForm-Noncommercial-1.0.0", "dependencies": { "@huggingface/transformers": "^3.0.0", diff --git a/gitnexus/package.json b/gitnexus/package.json index 26fdfce39..94f56946a 100644 --- a/gitnexus/package.json +++ b/gitnexus/package.json @@ -1,6 +1,6 @@ { "name": "gitnexus", - "version": "1.3.2", + "version": "1.3.3", "description": "Graph-powered code intelligence for AI agents. Index any codebase, query via MCP or CLI.", "author": "Abhigyan Patwari", "license": "PolyForm-Noncommercial-1.0.0", diff --git a/gitnexus/src/core/ingestion/filesystem-walker.ts b/gitnexus/src/core/ingestion/filesystem-walker.ts index c9365d3b0..7074593a0 100644 --- a/gitnexus/src/core/ingestion/filesystem-walker.ts +++ b/gitnexus/src/core/ingestion/filesystem-walker.ts @@ -21,8 +21,8 @@ export interface FilePath { const READ_CONCURRENCY = 32; -/** Skip files larger than 2MB — covers all real source files, excludes minified bundles/generated code */ -const MAX_FILE_SIZE = 2 * 1024 * 1024; +/** Skip files larger than 512KB — they're usually generated/vendored and crash tree-sitter */ +const MAX_FILE_SIZE = 512 * 1024; /** * Phase 1: Scan repository — stat files to get paths + sizes, no content loaded. @@ -69,7 +69,7 @@ export const walkRepositoryPaths = async ( } if (skippedLarge > 0) { - console.warn(` Skipped ${skippedLarge} files larger than ${Math.round(MAX_FILE_SIZE / 1024 / 1024)}MB (likely generated/vendored)`); + console.warn(` Skipped ${skippedLarge} large files (>${MAX_FILE_SIZE / 1024}KB, likely generated/vendored)`); } return entries; diff --git a/gitnexus/src/core/ingestion/parsing-processor.ts b/gitnexus/src/core/ingestion/parsing-processor.ts index 1134ebe49..c15d39a17 100644 --- a/gitnexus/src/core/ingestion/parsing-processor.ts +++ b/gitnexus/src/core/ingestion/parsing-processor.ts @@ -207,7 +207,7 @@ const processParsingSequential = async ( if (!language) continue; // Skip very large files — they can crash tree-sitter or cause OOM - if (file.content.length > 2 * 1024 * 1024) continue; + if (file.content.length > 512 * 1024) continue; await loadLanguage(language, file.path); diff --git a/gitnexus/src/core/ingestion/workers/parse-worker.ts b/gitnexus/src/core/ingestion/workers/parse-worker.ts index a89d8b402..2eee2c5bc 100644 --- a/gitnexus/src/core/ingestion/workers/parse-worker.ts +++ b/gitnexus/src/core/ingestion/workers/parse-worker.ts @@ -426,7 +426,7 @@ const processFileGroup = ( for (const file of files) { // Skip very large files — they can crash tree-sitter or cause OOM - if (file.content.length > 2 * 1024 * 1024) continue; + if (file.content.length > 512 * 1024) continue; let tree; try {