diff --git a/gitnexus/src/core/embeddings/embedding-pipeline.ts b/gitnexus/src/core/embeddings/embedding-pipeline.ts index cb1949144..cb28ca945 100644 --- a/gitnexus/src/core/embeddings/embedding-pipeline.ts +++ b/gitnexus/src/core/embeddings/embedding-pipeline.ts @@ -9,6 +9,7 @@ * 5. Create vector index for semantic search */ +import { createHash } from 'crypto'; import { initEmbedder, embedBatch, @@ -16,7 +17,7 @@ import { embeddingToArray, isEmbedderReady, } from './embedder.js'; -import { generateBatchEmbeddingTexts } from './text-generator.js'; +import { generateEmbeddingText, generateBatchEmbeddingTexts } from './text-generator.js'; import { type EmbeddingProgress, type EmbeddingConfig, @@ -26,9 +27,29 @@ import { DEFAULT_EMBEDDING_CONFIG, EMBEDDABLE_LABELS, } from './types.js'; +import { + EMBEDDING_TABLE_NAME, + EMBEDDING_INDEX_NAME, + CREATE_VECTOR_INDEX_QUERY, +} from '../lbug/schema.js'; +import { loadVectorExtension } from '../lbug/lbug-adapter.js'; const isDev = process.env.NODE_ENV === 'development'; +/** + * Compute a stable content fingerprint for an embeddable node. + * Used to detect when the underlying text has changed so stale vectors + * can be replaced (DELETE-then-INSERT, the Kuzu-sanctioned pattern for + * vector-indexed rows). + */ +export const contentHashForNode = ( + node: EmbeddableNode, + config: Partial = {}, +): string => { + const text = generateEmbeddingText(node, config); + return createHash('sha1').update(text).digest('hex'); +}; + /** * Progress callback type */ @@ -98,41 +119,32 @@ const batchInsertEmbeddings = async ( cypher: string, paramsList: Array>, ) => Promise, - updates: Array<{ id: string; embedding: number[] }>, + updates: Array<{ id: string; embedding: number[]; contentHash: string }>, ): Promise => { // MERGE instead of CREATE — idempotent, handles concurrent analyzes and partial prior runs - const cypher = `MERGE (e:CodeEmbedding {nodeId: $nodeId}) SET e.embedding = $embedding`; - const paramsList = updates.map((u) => ({ nodeId: u.id, embedding: u.embedding })); + const cypher = `MERGE (e:${EMBEDDING_TABLE_NAME} {nodeId: $nodeId}) SET e.embedding = $embedding, e.contentHash = $contentHash`; + const paramsList = updates.map((u) => ({ + nodeId: u.id, + embedding: u.embedding, + contentHash: u.contentHash, + })); await executeWithReusedStatement(cypher, paramsList); }; /** * Create the vector index for semantic search - * Now indexes the separate CodeEmbedding table + * Now indexes the separate CodeEmbedding table. + * Delegates extension loading to lbug-adapter's loadVectorExtension(), + * which owns the VECTOR extension lifecycle and state tracking. */ -let vectorExtensionLoaded = false; - const createVectorIndex = async ( executeQuery: (cypher: string) => Promise, ): Promise => { - // LadybugDB v0.15+ requires explicit VECTOR extension loading (once per session) - if (!vectorExtensionLoaded) { - try { - await executeQuery('INSTALL VECTOR'); - await executeQuery('LOAD EXTENSION VECTOR'); - vectorExtensionLoaded = true; - } catch { - // Extension may already be loaded — CREATE_VECTOR_INDEX will fail clearly if not - vectorExtensionLoaded = true; - } - } - - const cypher = ` - CALL CREATE_VECTOR_INDEX('CodeEmbedding', 'code_embedding_idx', 'embedding', metric := 'cosine') - `; + // Delegate to the adapter which tracks loaded state and handles DB reconnect resets + await loadVectorExtension(); try { - await executeQuery(cypher); + await executeQuery(CREATE_VECTOR_INDEX_QUERY); } catch (error) { // Index might already exist if (isDev) { @@ -148,7 +160,9 @@ const createVectorIndex = async ( * @param executeWithReusedStatement - Function to execute with reused prepared statement * @param onProgress - Callback for progress updates * @param config - Optional configuration override - * @param skipNodeIds - Optional set of node IDs that already have embeddings (incremental mode) + * @param existingEmbeddings - Optional map of nodeId → contentHash for incremental mode. + * Nodes whose hash matches are skipped; nodes with a changed hash are DELETE'd + * and re-embedded; nodes not in the map are embedded fresh. */ export const runEmbeddingPipeline = async ( executeQuery: (cypher: string) => Promise, @@ -158,7 +172,7 @@ export const runEmbeddingPipeline = async ( ) => Promise, onProgress: EmbeddingProgressCallback, config: Partial = {}, - skipNodeIds?: Set, + existingEmbeddings?: Map, ): Promise => { const finalConfig = { ...DEFAULT_EMBEDDING_CONFIG, ...config }; @@ -194,13 +208,57 @@ export const runEmbeddingPipeline = async ( // Phase 2: Query embeddable nodes let nodes = await queryEmbeddableNodes(executeQuery); - // Incremental mode: filter out nodes that already have embeddings - if (skipNodeIds && skipNodeIds.size > 0) { + // Incremental mode: compare content hashes, delete stale rows, skip fresh ones. + // Computed hashes for stale nodes are cached so batchInsertEmbeddings can reuse them + // (avoids double computation). + const computedStaleHashes = new Map(); + if (existingEmbeddings && existingEmbeddings.size > 0) { const beforeCount = nodes.length; - nodes = nodes.filter((n) => !skipNodeIds.has(n.id)); + const staleNodeIds: string[] = []; + nodes = nodes.filter((n) => { + const existingHash = existingEmbeddings.get(n.id); + if (existingHash === undefined) { + // New node — needs embedding + return true; + } + const currentHash = contentHashForNode(n, finalConfig); + if (currentHash !== existingHash) { + // Content changed — cache hash for reuse during insert, mark for DELETE + re-embed + computedStaleHashes.set(n.id, currentHash); + staleNodeIds.push(n.id); + return true; + } + // Hash matches — skip (fresh); no need to cache hash for skipped nodes + return false; + }); + + // DELETE stale embedding rows so they can be re-inserted + // (Kuzu forbids SET on vector-indexed properties; DELETE-then-INSERT is the sanctioned pattern) + if (staleNodeIds.length > 0) { + if (isDev) { + console.log(`🔄 Deleting ${staleNodeIds.length} stale embedding rows for re-embed`); + } + try { + await executeWithReusedStatement( + `MATCH (e:${EMBEDDING_TABLE_NAME} {nodeId: $nodeId}) DELETE e`, + staleNodeIds.map((nodeId) => ({ nodeId })), + ); + } catch (err) { + // "does not exist" = rows already gone — safe to proceed. + // All other errors risk vector-index corruption (Kuzu requires DELETE-before-INSERT + // for vector-indexed properties) — propagate so the pipeline aborts cleanly. + const msg = err instanceof Error ? err.message : String(err); + if (!msg.includes('does not exist')) { + throw new Error( + `[embed] Failed to delete stale embedding rows — aborting to prevent vector-index corruption: ${msg}`, + ); + } + } + } + if (isDev) { console.log( - `📦 Incremental embeddings: ${beforeCount} total, ${skipNodeIds.size} cached, ${nodes.length} to embed`, + `📦 Incremental embeddings: ${beforeCount} total, ${existingEmbeddings.size} cached, ${staleNodeIds.length} stale, ${nodes.length} to embed`, ); } } @@ -212,6 +270,11 @@ export const runEmbeddingPipeline = async ( } if (totalNodes === 0) { + // Ensure the vector index exists even when no new nodes need embedding. + // A prior crash or first-time incremental run may have left CodeEmbedding + // rows without ever reaching index creation. + await createVectorIndex(executeQuery); + onProgress({ phase: 'ready', percent: 100, @@ -250,6 +313,7 @@ export const runEmbeddingPipeline = async ( const updates = batch.map((node, i) => ({ id: node.id, embedding: embeddingToArray(embeddings[i]), + contentHash: computedStaleHashes.get(node.id) ?? contentHashForNode(node, finalConfig), })); await batchInsertEmbeddings(executeWithReusedStatement, updates); @@ -338,7 +402,7 @@ export const semanticSearch = async ( // Query the vector index on CodeEmbedding to get nodeIds and distances const vectorQuery = ` - CALL QUERY_VECTOR_INDEX('CodeEmbedding', 'code_embedding_idx', + CALL QUERY_VECTOR_INDEX('${EMBEDDING_TABLE_NAME}', '${EMBEDDING_INDEX_NAME}', CAST(${queryVecStr} AS FLOAT[${queryVec.length}]), ${k}) YIELD node AS emb, distance WITH emb, distance diff --git a/gitnexus/src/core/lbug/lbug-adapter.ts b/gitnexus/src/core/lbug/lbug-adapter.ts index 0298d5f7c..b5cd3e2ad 100644 --- a/gitnexus/src/core/lbug/lbug-adapter.ts +++ b/gitnexus/src/core/lbug/lbug-adapter.ts @@ -11,6 +11,7 @@ import { REL_TABLE_NAME, SCHEMA_QUERIES, EMBEDDING_TABLE_NAME, + STALE_HASH_SENTINEL, NodeTableName, } from './schema.js'; import { streamAllCSVsToDisk } from './csv-generator.js'; @@ -142,6 +143,16 @@ let currentDbPath: string | null = null; let ftsLoaded = false; let vectorExtensionLoaded = false; +/** + * Check if an error indicates a missing column or table (schema-level problem) + * rather than a transient/connection error. Used for legacy DB fallback logic. + */ +const isMissingColumnOrTableError = (msg: string): boolean => + msg.includes('does not exist') || + // Kuzu-specific: "(table|column|property) ... not found" — narrow enough to avoid + // matching transient errors like "connection not found" or "key not found". + /(table|column|property).*not found/i.test(msg); + /** Expose the current Database for pool adapter reuse in tests. */ export const getDatabase = (): lbug.Database | null => db; @@ -873,18 +884,35 @@ export const getLbugStats = async (): Promise<{ nodes: number; edges: number }> */ export const loadCachedEmbeddings = async (): Promise<{ embeddingNodeIds: Set; - embeddings: Array<{ nodeId: string; embedding: number[] }>; + embeddings: Array<{ nodeId: string; embedding: number[]; contentHash?: string }>; }> => { if (!conn) { return { embeddingNodeIds: new Set(), embeddings: [] }; } const embeddingNodeIds = new Set(); - const embeddings: Array<{ nodeId: string; embedding: number[] }> = []; + const embeddings: Array<{ nodeId: string; embedding: number[]; contentHash?: string }> = []; try { - const rows = await conn.query( - `MATCH (e:${EMBEDDING_TABLE_NAME}) RETURN e.nodeId AS nodeId, e.embedding AS embedding`, - ); + // Try to read contentHash alongside the embedding + let rows: any; + let hasContentHash = true; + try { + rows = await conn.query( + `MATCH (e:${EMBEDDING_TABLE_NAME}) RETURN e.nodeId AS nodeId, e.embedding AS embedding, e.contentHash AS contentHash`, + ); + } catch (err: any) { + // Only fall back for missing-column errors (legacy DBs without contentHash). + // Rethrow transient / connection errors so callers see them. + const msg = err?.message ?? ''; + if (isMissingColumnOrTableError(msg)) { + hasContentHash = false; + rows = await conn.query( + `MATCH (e:${EMBEDDING_TABLE_NAME}) RETURN e.nodeId AS nodeId, e.embedding AS embedding`, + ); + } else { + throw err; + } + } const result = Array.isArray(rows) ? rows[0] : rows; for (const row of await result.getAll()) { const nodeId = String(row.nodeId ?? row[0] ?? ''); @@ -897,6 +925,7 @@ export const loadCachedEmbeddings = async (): Promise<{ embedding: Array.isArray(embedding) ? embedding.map(Number) : Array.from(embedding as any).map(Number), + contentHash: hasContentHash ? (row.contentHash ?? row[2] ?? undefined) : undefined, }); } } @@ -907,6 +936,63 @@ export const loadCachedEmbeddings = async (): Promise<{ return { embeddingNodeIds, embeddings }; }; +/** + * Fetch existing embedding hashes from CodeEmbedding table for incremental embedding. + * Returns a Map suitable for passing to `runEmbeddingPipeline`. + * Handles legacy DBs without the `contentHash` column (all rows treated as stale with empty hash). + * Returns undefined if the CodeEmbedding table does not exist. + * + * @param execQuery - Cypher query executor (typically pool-adapter's `executeQuery`) + */ +export const fetchExistingEmbeddingHashes = async ( + execQuery: (cypher: string) => Promise, +): Promise | undefined> => { + try { + const rows = await execQuery( + `MATCH (e:${EMBEDDING_TABLE_NAME}) RETURN e.nodeId AS nodeId, e.contentHash AS contentHash`, + ); + if (!rows || rows.length === 0) return undefined; + const map = new Map(); + for (const r of rows) { + const nodeId = r.nodeId ?? r[0]; + const hash = r.contentHash ?? r[1] ?? STALE_HASH_SENTINEL; + if (nodeId) { + // Empty/null contentHash means legacy row — treat as stale so it gets re-embedded + map.set(nodeId, hash || STALE_HASH_SENTINEL); + } + } + return map; + } catch (err: any) { + const msg = err?.message ?? ''; + if (isMissingColumnOrTableError(msg)) { + // Column or table missing — try fallback without contentHash + try { + const rows = await execQuery(`MATCH (e:${EMBEDDING_TABLE_NAME}) RETURN e.nodeId AS nodeId`); + if (!rows || rows.length === 0) return undefined; + const map = new Map(); + for (const r of rows) { + const nodeId = r.nodeId ?? r[0]; + if (nodeId) map.set(nodeId, STALE_HASH_SENTINEL); // no contentHash — treat as stale + } + console.log( + `[embed] ${map.size} nodes in legacy DB (no contentHash) — all treated as stale`, + ); + return map; + } catch (fallbackErr: any) { + const fallbackMsg = fallbackErr?.message ?? ''; + if (isMissingColumnOrTableError(fallbackMsg)) { + console.log( + `[embed] CodeEmbedding table not yet present — full embedding run (${fallbackMsg})`, + ); + return undefined; + } + throw fallbackErr; + } + } + throw err; + } +}; + export const closeLbug = async (): Promise => { if (conn) { try { diff --git a/gitnexus/src/core/lbug/schema.ts b/gitnexus/src/core/lbug/schema.ts index 257938a01..c0ba10d4d 100644 --- a/gitnexus/src/core/lbug/schema.ts +++ b/gitnexus/src/core/lbug/schema.ts @@ -436,10 +436,20 @@ if (Number.isNaN(_rawDims) || _rawDims <= 0) { } export const EMBEDDING_DIMS = _rawDims; +/** HNSW vector index name for the CodeEmbedding table. */ +export const EMBEDDING_INDEX_NAME = 'code_embedding_idx'; + +/** + * Sentinel value for "no content hash available" — used in legacy DBs and null rows. + * Nodes with this hash are always treated as stale and re-embedded. + */ +export const STALE_HASH_SENTINEL = ''; + export const EMBEDDING_SCHEMA = ` CREATE NODE TABLE ${EMBEDDING_TABLE_NAME} ( nodeId STRING, embedding FLOAT[${EMBEDDING_DIMS}], + contentHash STRING, PRIMARY KEY (nodeId) )`; @@ -448,7 +458,7 @@ CREATE NODE TABLE ${EMBEDDING_TABLE_NAME} ( * Uses HNSW (Hierarchical Navigable Small World) algorithm with cosine similarity */ export const CREATE_VECTOR_INDEX_QUERY = ` -CALL CREATE_VECTOR_INDEX('${EMBEDDING_TABLE_NAME}', 'code_embedding_idx', 'embedding', metric := 'cosine') +CALL CREATE_VECTOR_INDEX('${EMBEDDING_TABLE_NAME}', '${EMBEDDING_INDEX_NAME}', 'embedding', metric := 'cosine') `; // ============================================================================ diff --git a/gitnexus/src/core/run-analyze.ts b/gitnexus/src/core/run-analyze.ts index 07fb8ab69..7c72fb592 100644 --- a/gitnexus/src/core/run-analyze.ts +++ b/gitnexus/src/core/run-analyze.ts @@ -32,6 +32,8 @@ import { } from '../storage/repo-manager.js'; import { getCurrentCommit, hasGitDir } from '../storage/git.js'; import { generateAIContextFiles } from '../cli/ai-context.js'; +import { EMBEDDING_TABLE_NAME } from './lbug/schema.js'; +import { STALE_HASH_SENTINEL } from './lbug/schema.js'; // --------------------------------------------------------------------------- // Public types @@ -138,7 +140,7 @@ export async function runFullAnalysis( // ── Cache embeddings from existing index before rebuild ──────────── let cachedEmbeddingNodeIds = new Set(); - let cachedEmbeddings: Array<{ nodeId: string; embedding: number[] }> = []; + let cachedEmbeddings: Array<{ nodeId: string; embedding: number[]; contentHash?: string }> = []; if (options.embeddings && existingMeta && !options.force) { try { @@ -219,10 +221,14 @@ export async function runFullAnalysis( const EMBED_BATCH = 200; for (let i = 0; i < cachedEmbeddings.length; i += EMBED_BATCH) { const batch = cachedEmbeddings.slice(i, i + EMBED_BATCH); - const paramsList = batch.map((e) => ({ nodeId: e.nodeId, embedding: e.embedding })); + const paramsList = batch.map((e) => ({ + nodeId: e.nodeId, + embedding: e.embedding, + contentHash: e.contentHash ?? STALE_HASH_SENTINEL, + })); try { await executeWithReusedStatement( - `MERGE (e:CodeEmbedding {nodeId: $nodeId}) SET e.embedding = $embedding`, + `MERGE (e:${EMBEDDING_TABLE_NAME} {nodeId: $nodeId}) SET e.embedding = $embedding, e.contentHash = $contentHash`, paramsList, ); } catch { @@ -251,6 +257,14 @@ export async function runFullAnalysis( httpMode ? 'Connecting to embedding endpoint...' : 'Loading embedding model...', ); const { runEmbeddingPipeline } = await import('./embeddings/embedding-pipeline.js'); + // Build a Map from cached embeddings for incremental mode + let existingEmbeddings: Map | undefined; + if (cachedEmbeddingNodeIds.size > 0) { + existingEmbeddings = new Map(); + for (const e of cachedEmbeddings) { + existingEmbeddings.set(e.nodeId, e.contentHash ?? STALE_HASH_SENTINEL); + } + } await runEmbeddingPipeline( executeQuery, executeWithReusedStatement, @@ -265,7 +279,7 @@ export async function runFullAnalysis( progress('embeddings', scaled, label); }, {}, - cachedEmbeddingNodeIds.size > 0 ? cachedEmbeddingNodeIds : undefined, + existingEmbeddings, ); } @@ -275,7 +289,9 @@ export async function runFullAnalysis( // Count embeddings in the index (cached + newly generated) let embeddingCount = 0; try { - const embResult = await executeQuery(`MATCH (e:CodeEmbedding) RETURN count(e) AS cnt`); + const embResult = await executeQuery( + `MATCH (e:${EMBEDDING_TABLE_NAME}) RETURN count(e) AS cnt`, + ); embeddingCount = embResult?.[0]?.cnt ?? 0; } catch { /* table may not exist if embeddings never ran */ diff --git a/gitnexus/src/server/api.ts b/gitnexus/src/server/api.ts index 9afdbfe0e..b0e47e019 100644 --- a/gitnexus/src/server/api.ts +++ b/gitnexus/src/server/api.ts @@ -1449,27 +1449,14 @@ export const createServer = async (port: number, host: string = '127.0.0.1') => await withLbugDb(lbugPath, async () => { const { runEmbeddingPipeline } = await import('../core/embeddings/embedding-pipeline.js'); - // Skip nodes that already have embeddings — Kuzu forbids SET on vector-indexed properties. - let skipNodeIds: Set | undefined; - try { - const rows = await executeQuery('MATCH (e:CodeEmbedding) RETURN e.nodeId AS nodeId'); - if (rows && rows.length > 0) { - skipNodeIds = new Set(rows.map((r: any) => r.nodeId ?? r[0]).filter(Boolean)); - console.log( - `[embed] ${skipNodeIds.size} nodes already embedded — skipping in incremental run`, - ); - } - } catch (err: any) { - // Swallow only "table does not exist" — let real connection errors propagate. - // Log so ops can see this path fire if Kuzu ever changes error wording. - const msg = err?.message ?? ''; - if (msg.includes('does not exist') || msg.includes('not found')) { - console.log( - `[embed] CodeEmbedding table not yet present — full embedding run (${msg})`, - ); - } else { - throw err; - } + // Fetch existing content hashes for incremental embedding. + // Delegated to lbug-adapter which owns the DB query logic and legacy-fallback handling. + const { fetchExistingEmbeddingHashes } = await import('../core/lbug/lbug-adapter.js'); + const existingEmbeddings = await fetchExistingEmbeddingHashes(executeQuery); + if (existingEmbeddings && existingEmbeddings.size > 0) { + console.log( + `[embed] ${existingEmbeddings.size} nodes already embedded — incremental run with content-hash comparison`, + ); } await runEmbeddingPipeline( executeQuery, @@ -1493,8 +1480,8 @@ export const createServer = async (port: number, host: string = '127.0.0.1') => }, }); }, - {}, // config: use defaults (runEmbeddingPipeline signature: executeQuery, executeWithReusedStatement, onProgress, config, skipNodeIds) - skipNodeIds, + {}, // config: use defaults + existingEmbeddings, ); }); diff --git a/gitnexus/test/unit/embedding-pipeline.test.ts b/gitnexus/test/unit/embedding-pipeline.test.ts new file mode 100644 index 000000000..7597d48ce --- /dev/null +++ b/gitnexus/test/unit/embedding-pipeline.test.ts @@ -0,0 +1,377 @@ +import { describe, it, expect, vi, beforeEach } from 'vitest'; +import { createHash } from 'crypto'; +import { contentHashForNode } from '../../src/core/embeddings/embedding-pipeline.js'; +import { generateEmbeddingText } from '../../src/core/embeddings/text-generator.js'; +import type { EmbeddableNode, EmbeddingProgress } from '../../src/core/embeddings/types.js'; +import { DEFAULT_EMBEDDING_CONFIG } from '../../src/core/embeddings/types.js'; +import { STALE_HASH_SENTINEL } from '../../src/core/lbug/schema.js'; + +// ──────────────────────────────────────────────────────────────────────────── +// contentHashForNode +// ──────────────────────────────────────────────────────────────────────────── +describe('contentHashForNode', () => { + const makeNode = (overrides: Partial = {}): EmbeddableNode => ({ + id: 'Function:foo:src/main.ts', + name: 'foo', + label: 'Function', + filePath: 'src/main.ts', + content: 'function foo() { return 1; }', + ...overrides, + }); + + it('returns a 40-char hex SHA-1 digest', () => { + const hash = contentHashForNode(makeNode()); + expect(hash).toMatch(/^[0-9a-f]{40}$/); + }); + + it('is deterministic — same node always produces the same hash', () => { + const node = makeNode(); + expect(contentHashForNode(node)).toBe(contentHashForNode(node)); + }); + + it('matches sha1(generateEmbeddingText(node))', () => { + const node = makeNode(); + const expected = createHash('sha1').update(generateEmbeddingText(node)).digest('hex'); + expect(contentHashForNode(node)).toBe(expected); + }); + + it('changes when node content is edited', () => { + const original = makeNode({ content: 'function foo() { return 1; }' }); + const edited = makeNode({ content: 'function foo() { return 42; }' }); + expect(contentHashForNode(original)).not.toBe(contentHashForNode(edited)); + }); + + it('changes when filePath differs', () => { + const a = makeNode({ filePath: 'src/a.ts' }); + const b = makeNode({ filePath: 'src/b.ts' }); + // Different filePaths lead to different embedding text ⇒ different hashes + expect(contentHashForNode(a)).not.toBe(contentHashForNode(b)); + }); + + it('produces identical hash regardless of config vs finalConfig when config is empty', () => { + const node = makeNode(); + const hashWithEmptyConfig = contentHashForNode(node, {}); + const hashWithFullDefaults = contentHashForNode(node, DEFAULT_EMBEDDING_CONFIG); + expect(hashWithEmptyConfig).toBe(hashWithFullDefaults); + }); +}); + +// ──────────────────────────────────────────────────────────────────────────── +// STALE_HASH_SENTINEL +// ──────────────────────────────────────────────────────────────────────────── +describe('STALE_HASH_SENTINEL', () => { + it('is the empty string', () => { + expect(STALE_HASH_SENTINEL).toBe(''); + }); + + it('is falsy — enables consistent `hash || STALE_HASH_SENTINEL` patterns', () => { + expect(!STALE_HASH_SENTINEL).toBe(true); + }); +}); + +// ──────────────────────────────────────────────────────────────────────────── +// runEmbeddingPipeline — exports +// ──────────────────────────────────────────────────────────────────────────── +describe('runEmbeddingPipeline incremental mode', () => { + it('exports contentHashForNode as a named export', async () => { + const mod = await import('../../src/core/embeddings/embedding-pipeline.js'); + expect(typeof mod.contentHashForNode).toBe('function'); + }); + + it('exports runEmbeddingPipeline as a named export', async () => { + const mod = await import('../../src/core/embeddings/embedding-pipeline.js'); + expect(typeof mod.runEmbeddingPipeline).toBe('function'); + }); +}); + +// ──────────────────────────────────────────────────────────────────────────── +// EMBEDDING_SCHEMA includes contentHash column +// ──────────────────────────────────────────────────────────────────────────── +describe('EMBEDDING_SCHEMA', () => { + it('includes contentHash STRING column', async () => { + const { EMBEDDING_SCHEMA } = await import('../../src/core/lbug/schema.js'); + expect(EMBEDDING_SCHEMA).toContain('contentHash STRING'); + }); +}); + +// ──────────────────────────────────────────────────────────────────────────── +// EMBEDDING_INDEX_NAME export +// ──────────────────────────────────────────────────────────────────────────── +describe('EMBEDDING_INDEX_NAME', () => { + it('is exported from schema.ts', async () => { + const { EMBEDDING_INDEX_NAME } = await import('../../src/core/lbug/schema.js'); + expect(EMBEDDING_INDEX_NAME).toBe('code_embedding_idx'); + }); +}); + +// ──────────────────────────────────────────────────────────────────────────── +// runEmbeddingPipeline — incremental filter logic with mocked embedder +// +// Tests the three incremental-mode code paths: +// 1. New node (not in existingEmbeddings) → embedded +// 2. Unchanged node (hash matches) → skipped +// 3. Stale node (hash mismatch) → DELETE old → re-embed +// 4. Zero nodes after filter → createVectorIndex still called +// ──────────────────────────────────────────────────────────────────────────── +describe('runEmbeddingPipeline incremental filter', () => { + // Track mocked calls + let queryCalls: string[]; + let stmtCalls: Array<{ cypher: string; params: Array> }>; + let progressUpdates: EmbeddingProgress[]; + + // Helper node + const makeNode = (overrides: Partial = {}): EmbeddableNode => ({ + id: 'Function:foo:src/main.ts', + name: 'foo', + label: 'Function', + filePath: 'src/main.ts', + content: 'function foo() { return 1; }', + ...overrides, + }); + + beforeEach(() => { + queryCalls = []; + stmtCalls = []; + progressUpdates = []; + vi.restoreAllMocks(); + vi.resetModules(); + }); + + // Mock the embedder module so we never need a real model + const mockEmbedderSetup = () => { + vi.doMock('../../src/core/embeddings/embedder.js', () => ({ + initEmbedder: vi.fn().mockResolvedValue(undefined), + embedBatch: vi + .fn() + .mockImplementation((texts: string[]) => + Promise.resolve(texts.map(() => new Float32Array(384))), + ), + embedText: vi.fn().mockResolvedValue(new Float32Array(384)), + embeddingToArray: vi.fn().mockImplementation((emb: Float32Array) => Array.from(emb)), + isEmbedderReady: vi.fn().mockReturnValue(true), + })); + + // Mock loadVectorExtension (avoids needing the native lbug module) + vi.doMock('../../src/core/lbug/lbug-adapter.js', () => ({ + loadVectorExtension: vi.fn().mockResolvedValue(undefined), + })); + }; + + const mockExecuteQuery = (nodes: EmbeddableNode[]) => { + return vi.fn().mockImplementation(async (cypher: string) => { + queryCalls.push(cypher); + // Respond to node queries based on label + for (const label of ['Function', 'Class', 'Method', 'Interface', 'File']) { + if (cypher.includes(`MATCH (n:${label})`)) { + return nodes + .filter((n) => n.label === label) + .map((n) => ({ + id: n.id, + name: n.name, + label: n.label, + filePath: n.filePath, + content: n.content, + startLine: n.startLine, + endLine: n.endLine, + })); + } + } + return []; + }); + }; + + const mockExecuteWithReusedStatement = () => { + return vi + .fn() + .mockImplementation(async (cypher: string, params: Array>) => { + stmtCalls.push({ cypher, params }); + }); + }; + + const onProgress = (p: EmbeddingProgress) => { + progressUpdates.push({ ...p }); + }; + + it('skips unchanged nodes when hash matches', async () => { + mockEmbedderSetup(); + + const node = makeNode(); + const hash = contentHashForNode(node, DEFAULT_EMBEDDING_CONFIG); + const existingEmbeddings = new Map([[node.id, hash]]); + + const executeQuery = mockExecuteQuery([node]); + const executeWithReusedStatement = mockExecuteWithReusedStatement(); + + const { runEmbeddingPipeline } = + await import('../../src/core/embeddings/embedding-pipeline.js'); + + await runEmbeddingPipeline( + executeQuery, + executeWithReusedStatement, + onProgress, + {}, + existingEmbeddings, + ); + + // No MERGE calls — node was skipped because hash matched + const mergeCalls = stmtCalls.filter((c) => c.cypher.includes('MERGE')); + expect(mergeCalls).toHaveLength(0); + + // Pipeline should reach 'ready' state + const readyProgress = progressUpdates.find((p) => p.phase === 'ready'); + expect(readyProgress).toBeDefined(); + expect(readyProgress!.percent).toBe(100); + }); + + it('embeds new nodes not in existingEmbeddings', async () => { + mockEmbedderSetup(); + + const node = makeNode({ + id: 'Function:newFn:src/new.ts', + name: 'newFn', + filePath: 'src/new.ts', + }); + const existingEmbeddings = new Map(); // empty — no prior embeddings + + const executeQuery = mockExecuteQuery([node]); + const executeWithReusedStatement = mockExecuteWithReusedStatement(); + + const { runEmbeddingPipeline } = + await import('../../src/core/embeddings/embedding-pipeline.js'); + + await runEmbeddingPipeline( + executeQuery, + executeWithReusedStatement, + onProgress, + {}, + existingEmbeddings, + ); + + // Should have a MERGE call to insert the embedding + const mergeCalls = stmtCalls.filter((c) => c.cypher.includes('MERGE')); + expect(mergeCalls.length).toBeGreaterThanOrEqual(1); + + // The inserted row should contain the node id and a contentHash + const insertParams = mergeCalls[0].params; + expect(insertParams.some((p: any) => p.nodeId === node.id)).toBe(true); + expect(insertParams[0].contentHash).toMatch(/^[0-9a-f]{40}$/); + }); + + it('deletes and re-embeds stale nodes (hash mismatch)', async () => { + mockEmbedderSetup(); + + const node = makeNode({ content: 'function foo() { return 42; }' }); + const staleHash = 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa'; // wrong hash + const existingEmbeddings = new Map([[node.id, staleHash]]); + + const executeQuery = mockExecuteQuery([node]); + const executeWithReusedStatement = mockExecuteWithReusedStatement(); + + const { runEmbeddingPipeline } = + await import('../../src/core/embeddings/embedding-pipeline.js'); + + await runEmbeddingPipeline( + executeQuery, + executeWithReusedStatement, + onProgress, + {}, + existingEmbeddings, + ); + + // Should have a DELETE call for the stale node + const deleteCalls = stmtCalls.filter((c) => c.cypher.includes('DELETE')); + expect(deleteCalls.length).toBeGreaterThanOrEqual(1); + expect(deleteCalls[0].params.some((p: any) => p.nodeId === node.id)).toBe(true); + + // Should also have a MERGE call to re-insert with new hash + const mergeCalls = stmtCalls.filter((c) => c.cypher.includes('MERGE')); + expect(mergeCalls.length).toBeGreaterThanOrEqual(1); + }); + + it('treats STALE_HASH_SENTINEL as stale — triggers re-embed', async () => { + mockEmbedderSetup(); + + const node = makeNode(); + // Legacy row: nodeId present but contentHash is STALE_HASH_SENTINEL + const existingEmbeddings = new Map([[node.id, STALE_HASH_SENTINEL]]); + + const executeQuery = mockExecuteQuery([node]); + const executeWithReusedStatement = mockExecuteWithReusedStatement(); + + const { runEmbeddingPipeline } = + await import('../../src/core/embeddings/embedding-pipeline.js'); + + await runEmbeddingPipeline( + executeQuery, + executeWithReusedStatement, + onProgress, + {}, + existingEmbeddings, + ); + + // Should have a DELETE call (stale) + const deleteCalls = stmtCalls.filter((c) => c.cypher.includes('DELETE')); + expect(deleteCalls.length).toBeGreaterThanOrEqual(1); + + // Should also have a MERGE (re-embed) + const mergeCalls = stmtCalls.filter((c) => c.cypher.includes('MERGE')); + expect(mergeCalls.length).toBeGreaterThanOrEqual(1); + }); + + it('calls createVectorIndex even when zero nodes need embedding after filter', async () => { + mockEmbedderSetup(); + + const node = makeNode(); + const hash = contentHashForNode(node, DEFAULT_EMBEDDING_CONFIG); + // All existing hashes match — zero nodes to embed + const existingEmbeddings = new Map([[node.id, hash]]); + + const executeQuery = mockExecuteQuery([node]); + const executeWithReusedStatement = mockExecuteWithReusedStatement(); + + const { runEmbeddingPipeline } = + await import('../../src/core/embeddings/embedding-pipeline.js'); + + await runEmbeddingPipeline( + executeQuery, + executeWithReusedStatement, + onProgress, + {}, + existingEmbeddings, + ); + + // The CREATE_VECTOR_INDEX query should have been called via executeQuery + const vectorIndexCalls = queryCalls.filter((c) => c.includes('CREATE_VECTOR_INDEX')); + expect(vectorIndexCalls.length).toBeGreaterThanOrEqual(1); + }); + + it('throws when DELETE for stale nodes fails with non-trivial error', async () => { + mockEmbedderSetup(); + + const node = makeNode({ content: 'function foo() { return 42; }' }); + const staleHash = 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa'; + const existingEmbeddings = new Map([[node.id, staleHash]]); + + const executeQuery = mockExecuteQuery([node]); + const executeWithReusedStatement = vi.fn().mockRejectedValue(new Error('Connection lost')); + + const { runEmbeddingPipeline } = + await import('../../src/core/embeddings/embedding-pipeline.js'); + + await expect( + runEmbeddingPipeline( + executeQuery, + executeWithReusedStatement, + onProgress, + {}, + existingEmbeddings, + ), + ).rejects.toThrow('vector-index corruption'); + }); +}); + +// ──────────────────────────────────────────────────────────────────────────── +// fetchExistingEmbeddingHashes — tested in integration tests (requires native module) +// The function is tested via lbug-core-adapter integration tests which have the +// native @ladybugdb/core module available. +// ────────────────────────────────────────────────────────────────────────────