fix: content-hash staleness detection for embeddings and vector index creation on zero-node path (#831)

* Initial plan * fix: stale vectors preserved on content edits and vector index missing after zero-node run Issue 1: Add contentHash to EMBEDDING_SCHEMA and embedding pipeline. - contentHash column persisted per CodeEmbedding row - POST /api/embed queries nodeId+contentHash, compares per-node hash - Stale rows (hash mismatch) are DELETE'd before re-embedding - Legacy DBs without contentHash treated as stale (full re-embed) - loadCachedEmbeddings and run-analyze cache restore include contentHash Issue 2: createVectorIndex called unconditionally before zero-node early return. Regression tests: - contentHashForNode determinism and content-change detection - EMBEDDING_SCHEMA includes contentHash STRING column - Pipeline exports verified Agent-Logs-Url: https://github.com/abhigyanpatwari/GitNexus/sessions/1581c0c0-f359-4376-b47e-62d24a28fd2d Co-authored-by: magyargergo <11230420+magyargergo@users.noreply.github.com> * fix: use parameterized query for stale embedding DELETE, revert package-lock.json Agent-Logs-Url: https://github.com/abhigyanpatwari/GitNexus/sessions/1581c0c0-f359-4376-b47e-62d24a28fd2d Co-authored-by: magyargergo <11230420+magyargergo@users.noreply.github.com> * fix: address review feedback — config consistency, narrow catches, extract DB logic Bug #1: Use finalConfig consistently in contentHashForNode (line 224 was using raw `config` while line 307 used `finalConfig`). Cache precomputed hashes in filter phase to avoid double computation (Perf #5). Bug #2: Narrow catch in loadCachedEmbeddings to only fall back on column/table-missing errors. Rethrow transient/connection errors. Bug #3: Log non-trivial DELETE failures instead of silently swallowing. Arch Violation #3: Extract fetchExistingEmbeddingHashes from api.ts into lbug-adapter.ts. Server layer now calls a single adapter function instead of re-implementing the DB query logic with nested try-catch. Tests: Add config consistency test, note that fetchExistingEmbeddingHashes tests require native module (run in CI). Agent-Logs-Url: https://github.com/abhigyanpatwari/GitNexus/sessions/b8c4f6b0-4095-4507-a15d-d8469793efac Co-authored-by: magyargergo <11230420+magyargergo@users.noreply.github.com> * fix: narrow Column error match to 'contentHash' in lbug-adapter fallback checks Agent-Logs-Url: https://github.com/abhigyanpatwari/GitNexus/sessions/b8c4f6b0-4095-4507-a15d-d8469793efac Co-authored-by: magyargergo <11230420+magyargergo@users.noreply.github.com> * fix: address production-readiness review — eliminate competing state, use schema constants, hard-fail on stale DELETE, add incremental filter tests Gap A / Arch Violation 1: Remove duplicate vectorExtensionLoaded flag from embedding-pipeline.ts — delegate to lbug-adapter's loadVectorExtension() which owns the VECTOR extension lifecycle and resets on DB reconnect. Arch Violation 2: Replace all hardcoded 'CodeEmbedding' and 'code_embedding_idx' strings in embedding-pipeline.ts and run-analyze.ts with EMBEDDING_TABLE_NAME, EMBEDDING_INDEX_NAME, and CREATE_VECTOR_INDEX_QUERY imported from schema.ts. Add EMBEDDING_INDEX_NAME export to schema.ts. Gap B: Make DELETE failure for stale vectors a hard throw (not just a warning). Continuing after failed DELETE risks Kuzu vector-index corruption since the constraint requires DELETE-before-INSERT for vector-indexed properties. "not found" / "does not exist" errors are still safe to ignore. STALE_HASH_SENTINEL: Define a named constant in embedding types.ts for the empty-string sentinel convention. Used consistently in lbug-adapter.ts and run-analyze.ts so the invariant is self-documenting. Tests: Add comprehensive unit tests for the incremental filter logic with mocked embedder: - New node → embedded - Unchanged node (hash matches) → skipped - Stale node (hash mismatch) → DELETE + re-embed - STALE_HASH_SENTINEL → treated as stale - Zero nodes after filter → createVectorIndex still called - DELETE failure with non-trivial error → throws Agent-Logs-Url: https://github.com/abhigyanpatwari/GitNexus/sessions/b21edee7-c9c5-4742-947b-d0def4fb26aa Co-authored-by: magyargergo <11230420+magyargergo@users.noreply.github.com> * fix: tighten error classification — extract isMissingColumnOrTableError helper, remove broad pattern matching - Extract isMissingColumnOrTableError() helper in lbug-adapter for consistent schema-error detection (replaces duplicate inline checks) - Tighten 'contentHash' match: now requires 'property' AND 'contentHash' (Kuzu-specific pattern) instead of broad 'contentHash' substring - Tighten DELETE error check: only ignore 'does not exist' (Kuzu's actual message), not broad 'not found' which could mask connection errors - Fix test node ID/name/filePath consistency Agent-Logs-Url: https://github.com/abhigyanpatwari/GitNexus/sessions/b21edee7-c9c5-4742-947b-d0def4fb26aa Co-authored-by: magyargergo <11230420+magyargergo@users.noreply.github.com> * fix: CI failures and final review — move STALE_HASH_SENTINEL to schema, tighten error matching, fix test mocking, format - Move STALE_HASH_SENTINEL from embeddings/types.ts to lbug/schema.ts (fixes inverted layer dependency: lbug should not import from embeddings) - Tighten isMissingColumnOrTableError: replace broad msg.includes('not found') with /(table|column|property).*not found/i regex to avoid matching transient errors - Add vi.resetModules() in test beforeEach for explicit module isolation (fixes vi.doMock not intercepting loadVectorExtension in CI) - Skip precomputedHashes.set() on unchanged (return false) path - Run prettier on all 5 files flagged by CI format check Agent-Logs-Url: https://github.com/abhigyanpatwari/GitNexus/sessions/e20311fd-4361-47b4-a137-9adc3e533b35 * fix: address remaining review nits — rename precomputedHashes, generalize error matcher, revert package-lock - Rename precomputedHashes → computedStaleHashes (hashes are computed on-demand during filter, only cached for stale nodes being re-embedded) - Remove contentHash-specific clause from isMissingColumnOrTableError — the regex /(table|column|property).*not found/i already covers it - Revert package-lock.json ssh→https protocol change Agent-Logs-Url: https://github.com/abhigyanpatwari/GitNexus/sessions/e20311fd-4361-47b4-a137-9adc3e533b35 --------- Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com> Co-authored-by: magyargergo <11230420+magyargergo@users.noreply.github.com>
2026-09-11 22:53:04 +00:00 · 2026-04-15 11:20:48 +01:00 · 2026-04-15 11:20:48 +01:00 · 1df79c2eab
commit 1df79c2eab
parent 32c9ddaf32
6 changed files with 604 additions and 64 deletions
--- 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<EmbeddingConfig> = {},
+): 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<Record<string, any>>,
  ) => Promise<void>,
-  updates: Array<{ id: string; embedding: number[] }>,
+  updates: Array<{ id: string; embedding: number[]; contentHash: string }>,
 ): Promise<void> => {
  // 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<any[]>,
 ): Promise<void> => {
-  // 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<any[]>,
@ -158,7 +172,7 @@ export const runEmbeddingPipeline = async (
  ) => Promise<void>,
  onProgress: EmbeddingProgressCallback,
  config: Partial<EmbeddingConfig> = {},
-  skipNodeIds?: Set<string>,
+  existingEmbeddings?: Map<string, string>,
 ): Promise<void> => {
  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<string, string>();
+    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
--- 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<string>;
-  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<string>();
-  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<nodeId, contentHash> 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<any[]>,
+): Promise<Map<string, string> | 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<string, string>();
+    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<string, string>();
+        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<void> => {
  if (conn) {
    try {
--- 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')
 `;

 // ============================================================================
--- 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<string>();
-  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<nodeId, contentHash> from cached embeddings for incremental mode
+      let existingEmbeddings: Map<string, string> | undefined;
+      if (cachedEmbeddingNodeIds.size > 0) {
+        existingEmbeddings = new Map<string, string>();
+        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 */
--- 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<string> | 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,
            );
          });

--- a/gitnexus/test/unit/embedding-pipeline.test.ts
+++ 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> = {}): 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<Record<string, any>> }>;
+  let progressUpdates: EmbeddingProgress[];
+
+  // Helper node
+  const makeNode = (overrides: Partial<EmbeddableNode> = {}): 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<Record<string, any>>) => {
+        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<string, string>([[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<string, string>(); // 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<string, string>([[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<string, string>([[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<string, string>([[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<string, string>([[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.
+// ────────────────────────────────────────────────────────────────────────────