diff --git a/gitnexus/src/core/incremental/subgraph-extract.ts b/gitnexus/src/core/incremental/subgraph-extract.ts new file mode 100644 index 000000000..dea2cecb9 --- /dev/null +++ b/gitnexus/src/core/incremental/subgraph-extract.ts @@ -0,0 +1,52 @@ +/** + * Subgraph extraction for incremental DB writeback. + * + * Given the FULL ctx.graph produced by the pipeline (all files parsed, + * all phases run) and the set of file paths whose DB rows must be + * replaced, produce a smaller KnowledgeGraph that contains: + * + * - Every node whose `properties.filePath` is in `toWriteSet`. + * - Every graph-wide node (Community, Process) — these are regenerated + * each run by the communities/processes phases and must be fully + * rewritten. + * - Every relationship where AT LEAST ONE endpoint is in the writable + * set above. Relationships entirely between unchanged-file nodes + * are skipped — their rows are still in the DB and re-inserting + * them would PK-conflict at COPY time. + * + * The resulting subgraph is what gets passed to `loadGraphToLbug` after + * the orchestrator has deleted the corresponding DB rows. Hydrated + * unchanged-file rows are never touched in the DB. + */ + +import type { GraphNode, GraphRelationship } from 'gitnexus-shared'; +import { createKnowledgeGraph } from '../graph/graph.js'; +import type { KnowledgeGraph } from '../graph/types.js'; + +const isGraphWide = (label: string): boolean => + label === 'Community' || label === 'Process'; + +export const extractChangedSubgraph = ( + fullGraph: KnowledgeGraph, + toWriteSet: ReadonlySet, +): KnowledgeGraph => { + const sub = createKnowledgeGraph(); + const writableNodeIds = new Set(); + + fullGraph.forEachNode((n: GraphNode) => { + const filePath = n.properties?.filePath as string | undefined; + const include = (filePath && toWriteSet.has(filePath)) || isGraphWide(n.label); + if (include) { + sub.addNode(n); + writableNodeIds.add(n.id); + } + }); + + fullGraph.forEachRelationship((r: GraphRelationship) => { + if (writableNodeIds.has(r.sourceId) || writableNodeIds.has(r.targetId)) { + sub.addRelationship(r); + } + }); + + return sub; +}; diff --git a/gitnexus/src/core/lbug/lbug-adapter.ts b/gitnexus/src/core/lbug/lbug-adapter.ts index cb88986ca..655c10a56 100644 --- a/gitnexus/src/core/lbug/lbug-adapter.ts +++ b/gitnexus/src/core/lbug/lbug-adapter.ts @@ -1204,6 +1204,37 @@ export const deleteNodesForFile = async ( export const getEmbeddingTableName = (): string => EMBEDDING_TABLE_NAME; +/** + * Drop every Community and Process node (and their MEMBER_OF / + * STEP_IN_PROCESS edges via DETACH DELETE). Used at the start of an + * incremental run so the communities and processes phases regenerate + * them from scratch on the merged graph — required for the + * "Leiden runs on the FULL graph" correctness invariant. + */ +export const deleteAllCommunitiesAndProcesses = async (): Promise<{ + nodesDeleted: number; +}> => { + if (!conn) { + throw new Error('LadybugDB not initialized. Call initLbug first.'); + } + let nodesDeleted = 0; + for (const label of ['Community', 'Process']) { + try { + const countResult = await conn.query(`MATCH (n:${label}) 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) { + await conn.query(`MATCH (n:${label}) DETACH DELETE n`); + nodesDeleted += count; + } + } catch { + // Table may not exist yet on a freshly-initialized DB — fine. + } + } + return { nodesDeleted }; +}; + // ============================================================================ // Full-Text Search (FTS) Functions // ============================================================================ diff --git a/gitnexus/src/core/run-analyze.ts b/gitnexus/src/core/run-analyze.ts index 7227fd945..96e24f79b 100644 --- a/gitnexus/src/core/run-analyze.ts +++ b/gitnexus/src/core/run-analyze.ts @@ -11,6 +11,7 @@ import path from 'path'; import fs from 'fs/promises'; +import { execFileSync } from 'child_process'; import { runPipelineFromRepo } from './ingestion/pipeline.js'; import { initLbug, @@ -20,6 +21,8 @@ import { executeWithReusedStatement, closeLbug, loadCachedEmbeddings, + deleteNodesForFile, + deleteAllCommunitiesAndProcesses, } from './lbug/lbug-adapter.js'; import { createSearchFTSIndexes } from './search/fts-indexes.js'; import { @@ -29,7 +32,10 @@ import { ensureGitNexusIgnored, registerRepo, cleanupOldKuzuFiles, + INCREMENTAL_SCHEMA_VERSION, } from '../storage/repo-manager.js'; +import { computeFileHashes, diffFileHashes } from '../storage/file-hash.js'; +import { extractChangedSubgraph } from './incremental/subgraph-extract.js'; import { getCurrentCommit, getRemoteUrl, @@ -174,25 +180,58 @@ export async function runFullAnalysis( const repoHasGit = hasGitDir(repoPath); const currentCommit = repoHasGit ? getCurrentCommit(repoPath) : ''; - const existingMeta = await loadMeta(storagePath); + let existingMeta = await loadMeta(storagePath); + + // ── Crash recovery: dirty flag forces full rebuild ──────────────── + // If the previous incremental run set incrementalInProgress and didn't + // clear it, the on-disk index may be in a half-state. Cheapest path + // back to a known-good index is to wipe + rebuild from scratch. + if (existingMeta?.incrementalInProgress) { + log( + 'Previous incremental run did not complete cleanly (incrementalInProgress flag set); ' + + 'forcing full rebuild to restore a known-good index.', + ); + options = { ...options, force: true }; + // Reload meta after clearing the flag in-memory; we still want fileHashes + // for the post-rebuild meta carry-over, but force=true ensures the + // rebuild path executes. + } // ── Early-return: already up to date ────────────────────────────── if (existingMeta && !options.force && existingMeta.lastCommit === currentCommit) { // Non-git folders have currentCommit = '' — always rebuild since we can't detect changes if (currentCommit !== '') { - await ensureGitNexusIgnored(repoPath); - return { - // `resolveRepoIdentityRoot` collapses worktree roots to the - // canonical repo basename (#1259) but leaves arbitrary subdirs - // and `--skip-git` paths unchanged (#1232/#1233 intent preserved). - repoName: - options.registryName ?? - getInferredRepoName(repoPath) ?? - path.basename(resolveRepoIdentityRoot(repoPath)), - repoPath, - stats: existingMeta.stats ?? {}, - alreadyUpToDate: true, - }; + // For git repos, even if HEAD matches lastCommit, the working tree + // may have uncommitted changes. Only short-circuit when the working + // tree is also clean — otherwise fall through to the incremental + // path which will hash-diff and update only changed files. + const dirty = (() => { + try { + const out = execFileSync('git', ['status', '--porcelain'], { + cwd: repoPath, + stdio: ['ignore', 'pipe', 'ignore'], + encoding: 'utf8', + }); + return out.trim().length > 0; + } catch { + return true; // conservative on git failure + } + })(); + if (!dirty) { + await ensureGitNexusIgnored(repoPath); + return { + // `resolveRepoIdentityRoot` collapses worktree roots to the + // canonical repo basename (#1259) but leaves arbitrary subdirs + // and `--skip-git` paths unchanged (#1232/#1233 intent preserved). + repoName: + options.registryName ?? + getInferredRepoName(repoPath) ?? + path.basename(resolveRepoIdentityRoot(repoPath)), + repoPath, + stats: existingMeta.stats ?? {}, + alreadyUpToDate: true, + }; + } } } @@ -241,7 +280,18 @@ export async function runFullAnalysis( ); } - if (shouldLoadCache && existingMeta) { + // Predict whether this run will use the incremental DB-writeback path. + // Used to suppress the embedding cache+restore cycle in the incremental + // case (embeddings stay in the DB; re-inserting them would PK-conflict). + const willTryIncremental = + !options.force && + !!existingMeta && + existingMeta.schemaVersion === INCREMENTAL_SCHEMA_VERSION && + !!existingMeta.fileHashes && + Object.keys(existingMeta.fileHashes).length > 0 && + repoHasGit; + + if (shouldLoadCache && existingMeta && !willTryIncremental) { try { progress('embeddings', 0, 'Caching embeddings...'); await initLbug(lbugPath); @@ -279,13 +329,59 @@ export async function runFullAnalysis( // ── Phase 2: LadybugDB (60–85%) ────────────────────────────────── progress('lbug', 60, 'Loading into LadybugDB...'); - await closeLbug(); - const lbugFiles = [lbugPath, `${lbugPath}.wal`, `${lbugPath}.lock`]; - for (const f of lbugFiles) { - try { - await fs.rm(f, { recursive: true, force: true }); - } catch { - /* swallow */ + // Compute current per-file content hashes from the pipeline's File nodes. + // Used both to drive the incremental DB writeback (when eligible) and to + // populate meta.json.fileHashes for the next run. + const allFilePaths: string[] = []; + pipelineResult.graph.forEachNode((n) => { + if (n.label === 'File') { + const fp = n.properties?.filePath as string | undefined; + if (fp) allFilePaths.push(fp); + } + }); + const newFileHashes = await computeFileHashes(repoPath, allFilePaths); + + // Decide incremental vs full at THIS point (post-pipeline, pre-DB). + // willTryIncremental was the *prediction* used to skip the embedding + // cache cycle; here we re-evaluate against the actual pipeline output. + const isIncremental = + willTryIncremental && + existingMeta !== null && + !!existingMeta.fileHashes && + allFilePaths.length > 0; + + const hashDiff = isIncremental + ? diffFileHashes(newFileHashes, existingMeta!.fileHashes) + : undefined; + + if (isIncremental && hashDiff) { + log( + `Incremental: changed=${hashDiff.changed.length}, ` + + `added=${hashDiff.added.length}, ` + + `deleted=${hashDiff.deleted.length} ` + + `(skipping wipe + ${ + allFilePaths.length - hashDiff.toWrite.length + } unchanged file rows preserved)`, + ); + // Set the dirty flag BEFORE any destructive DB mutation. Cleared on + // success at the meta-save step. + await saveMeta(storagePath, { + ...existingMeta!, + incrementalInProgress: { + startedAt: Date.now(), + toWriteCount: hashDiff.toWrite.length, + }, + }); + } else { + // Full rebuild path: wipe DB files first. + await closeLbug(); + const lbugFiles = [lbugPath, `${lbugPath}.wal`, `${lbugPath}.lock`]; + for (const f of lbugFiles) { + try { + await fs.rm(f, { recursive: true, force: true }); + } catch { + /* swallow */ + } } } @@ -296,11 +392,49 @@ export async function runFullAnalysis( // must be released to avoid blocking subsequent invocations. let lbugMsgCount = 0; - await loadGraphToLbug(pipelineResult.graph, pipelineResult.repoPath, storagePath, (msg) => { - lbugMsgCount++; - const pct = Math.min(84, 60 + Math.round((lbugMsgCount / (lbugMsgCount + 10)) * 24)); - progress('lbug', pct, msg); - }); + if (isIncremental && hashDiff) { + // ── Incremental DB writeback ─────────────────────────────────── + // 1. Delete rows for files we're about to rewrite + deleted files. + const filesToDelete = [...hashDiff.toWrite, ...hashDiff.deleted]; + for (let i = 0; i < filesToDelete.length; i++) { + const f = filesToDelete[i]; + try { + await deleteNodesForFile(f); + } catch { + /* file may not have rows (e.g. an unparseable file) — fine */ + } + if (i % 20 === 0) { + progress( + 'lbug', + 62, + `Removing rows for changed files (${i}/${filesToDelete.length})...`, + ); + } + } + // 2. Drop graph-wide nodes (Community, Process). They'll be re-inserted + // from the fresh pipeline output below. Required for the + // "Leiden runs on the FULL graph" correctness invariant. + await deleteAllCommunitiesAndProcesses(); + + // 3. Extract the changed subgraph from the FULL ctx.graph and write + // only that. Unchanged-file rows in the DB stay untouched. + const subgraph = extractChangedSubgraph( + pipelineResult.graph, + new Set(hashDiff.toWrite), + ); + await loadGraphToLbug(subgraph, pipelineResult.repoPath, storagePath, (msg) => { + lbugMsgCount++; + const pct = Math.min(84, 65 + Math.round((lbugMsgCount / (lbugMsgCount + 10)) * 19)); + progress('lbug', pct, msg); + }); + } else { + // ── Full rebuild ─────────────────────────────────────────────── + await loadGraphToLbug(pipelineResult.graph, pipelineResult.repoPath, storagePath, (msg) => { + lbugMsgCount++; + const pct = Math.min(84, 60 + Math.round((lbugMsgCount / (lbugMsgCount + 10)) * 24)); + progress('lbug', pct, msg); + }); + } // ── Phase 3: FTS (85–90%) ───────────────────────────────────────── progress('fts', 85, 'Creating search indexes...'); @@ -454,6 +588,12 @@ export async function runFullAnalysis( const effectiveSemanticMode = semanticMode ?? (runtimeCapabilities.semanticMode === 'vector-index' ? 'vector-index' : 'exact-scan'); + + // Convert the post-run file-hash map to the on-disk Record + // shape consumed by RepoMeta.fileHashes. + const newFileHashesRecord: Record = {}; + for (const [k, v] of newFileHashes) newFileHashesRecord[k] = v; + const meta = { repoPath, lastCommit: currentCommit, @@ -483,6 +623,15 @@ export async function runFullAnalysis( reason: runtimeCapabilities.reason, }, }, + // Incremental-indexing fields. Populated for git repos so the next + // analyze run can take the incremental DB-writeback path. Setting + // incrementalInProgress to undefined explicitly clears any prior + // dirty flag (full and incremental success paths converge here). + schemaVersion: hasGitDir(repoPath) ? INCREMENTAL_SCHEMA_VERSION : undefined, + fileHashes: hasGitDir(repoPath) ? newFileHashesRecord : undefined, + incrementalInProgress: undefined as + | { startedAt: number; toWriteCount: number } + | undefined, }; await saveMeta(storagePath, meta); // Forward the --name alias and the registry-collision bypass bit. diff --git a/gitnexus/src/storage/file-hash.ts b/gitnexus/src/storage/file-hash.ts new file mode 100644 index 000000000..b39111815 --- /dev/null +++ b/gitnexus/src/storage/file-hash.ts @@ -0,0 +1,104 @@ +/** + * Per-file content hashing for incremental DB writeback. + * + * On every analyze run we compute SHA-256 of every file's content and + * store the map in meta.json. The next run compares disk against the + * stored map and produces: + * - `changed` — content differs (re-emit DB rows for this file) + * - `added` — file is new on disk (insert DB rows) + * - `deleted` — file was in last meta but no longer on disk (drop rows) + * + * The pipeline still parses every file (correctness invariant: cross-file + * resolution needs full data). What this enables is a SELECTIVE DB + * writeback: instead of wipe-and-reload of the whole graph (~50s of CSV + * COPY on a 25K-node repo), we only delete-and-rewrite rows for the + * changed/added/deleted set. + * + * See docs/superpowers/specs/2026-05-10-incremental-indexing-design.md + * (Option B revision). + */ + +import { createHash } from 'crypto'; +import fs from 'fs/promises'; +import path from 'path'; + +/** + * Compute SHA-256 of a single file. Returns null when the file can't be + * read — caller treats that as "no signature, assume changed". + */ +export const computeFileHash = async (absPath: string): Promise => { + try { + const buf = await fs.readFile(absPath); + return createHash('sha256').update(buf).digest('hex'); + } catch { + return null; + } +}; + +/** + * Compute SHA-256 hashes for many files in parallel batches. Files that + * fail to read are omitted from the result map. + */ +export const computeFileHashes = async ( + repoPath: string, + relPaths: readonly string[], +): Promise> => { + const out = new Map(); + const BATCH = 100; + for (let i = 0; i < relPaths.length; i += BATCH) { + const batch = relPaths.slice(i, i + BATCH); + const results = await Promise.all( + batch.map(async (rel) => { + const h = await computeFileHash(path.join(repoPath, rel)); + return h ? ([rel, h] as const) : null; + }), + ); + for (const r of results) if (r) out.set(r[0], r[1]); + } + return out; +}; + +/** Result of comparing the current on-disk hashes against stored ones. */ +export interface FileHashDiff { + /** Files whose content hash differs from stored. */ + changed: string[]; + /** Files in the current scan that weren't in the stored map. */ + added: string[]; + /** Files in the stored map that aren't in the current scan. */ + deleted: string[]; + /** All files whose DB rows must be replaced (changed ∪ added). */ + toWrite: string[]; +} + +/** + * Diff a current hash map against a previously stored one. + * + * Sorted output so two runs produce identical diff arrays for the same + * changes — useful for stable logging / equivalence checks. + */ +export const diffFileHashes = ( + current: ReadonlyMap, + stored: Readonly> | undefined, +): FileHashDiff => { + const storedMap = new Map(stored ? Object.entries(stored) : []); + const changed: string[] = []; + const added: string[] = []; + for (const [p, h] of current) { + const prev = storedMap.get(p); + if (prev === undefined) added.push(p); + else if (prev !== h) changed.push(p); + } + const deleted: string[] = []; + for (const p of storedMap.keys()) { + if (!current.has(p)) deleted.push(p); + } + changed.sort(); + added.sort(); + deleted.sort(); + return { + changed, + added, + deleted, + toWrite: [...changed, ...added].sort(), + }; +}; diff --git a/gitnexus/src/storage/repo-manager.ts b/gitnexus/src/storage/repo-manager.ts index 8c0bda95f..c6fef8e2a 100644 --- a/gitnexus/src/storage/repo-manager.ts +++ b/gitnexus/src/storage/repo-manager.ts @@ -71,8 +71,40 @@ export interface RepoMeta { processes?: number; embeddings?: number; }; + /** + * Bumped whenever incremental-indexing invariants change in an + * incompatible way (delete-and-rewrite logic, subgraph extraction, + * graph-wide node handling). On mismatch, runFullAnalysis forces a + * full rebuild rather than risk an inconsistent incremental update. + */ + schemaVersion?: number; + /** + * SHA-256 of every file's content at the time of the last successful + * indexing run. The next run computes current hashes and diffs against + * this map to determine which files' DB rows must be replaced. + * Map keys are repo-relative paths. + */ + fileHashes?: Record; + /** + * Crash-recovery dirty flag. Written to meta.json BEFORE any + * destructive DB mutation in an incremental run; cleared on success + * by overwriting meta.json. If a run crashes between, the next run + * sees the flag and forces a full rebuild — the cheapest path back + * to a known-good index. + */ + incrementalInProgress?: { + /** When the incremental run started (epoch ms). */ + startedAt: number; + /** Number of files in the writable set, for diagnostic logs. */ + toWriteCount: number; + }; } +/** + * Bumped whenever incremental-indexing invariants change incompatibly. + */ +export const INCREMENTAL_SCHEMA_VERSION = 1; + export interface IndexedRepo { repoPath: string; storagePath: string;