From ea1f59a233a712ceaa995453537bc5b62ce98415 Mon Sep 17 00:00:00 2001 From: Eva Date: Mon, 13 Jul 2026 21:49:45 +0700 Subject: [PATCH 1/2] fix(vector): load extension before indexed embedding deletes --- gitnexus/src/core/lbug/lbug-adapter.ts | 5 +++ .../integration/lbug-vector-extension.test.ts | 31 +++++++++++++++++++ 2 files changed, 36 insertions(+) diff --git a/gitnexus/src/core/lbug/lbug-adapter.ts b/gitnexus/src/core/lbug/lbug-adapter.ts index 83b13d5f95..573154812b 100644 --- a/gitnexus/src/core/lbug/lbug-adapter.ts +++ b/gitnexus/src/core/lbug/lbug-adapter.ts @@ -2294,6 +2294,11 @@ export const deleteNodesForFiles = async ( if (!conn) { throw new Error('LadybugDB not initialized. Call initLbug first.'); } + // A persisted HNSW index makes even ordinary CodeEmbedding DELETEs depend + // on VECTOR in every new connection. Keep this mutation path offline-only: + // if VECTOR is unavailable, the delete below still works for an unindexed + // table and surfaces LadybugDB's real error for an indexed one. + await loadVectorExtension(undefined, { policy: 'load-only' }); const targetConn = conn; let warnedMissingEmbeddingTable = false; for (let i = 0; i < filePaths.length; i += DELETE_FILES_CHUNK_SIZE) { diff --git a/gitnexus/test/integration/lbug-vector-extension.test.ts b/gitnexus/test/integration/lbug-vector-extension.test.ts index 5d0e18e013..17d03a26d7 100644 --- a/gitnexus/test/integration/lbug-vector-extension.test.ts +++ b/gitnexus/test/integration/lbug-vector-extension.test.ts @@ -154,6 +154,37 @@ withTestLbugDB('vector-index-creation', (handle) => { const rows = await adapter.executeQuery('CALL SHOW_INDEXES() RETURN *'); expect(rows.filter((row: any) => row.index_name === 'code_embedding_idx')).toHaveLength(1); }); + + it('deletes embedding rows from a persisted HNSW index after close and re-open', async () => { + const adapter = await import('../../src/core/lbug/lbug-adapter.js'); + const { batchInsertEmbeddings } = + await import('../../src/core/embeddings/embedding-pipeline.js'); + const { EMBEDDING_DIMS } = await import('../../src/core/lbug/schema.js'); + const filePath = 'src/vector-reopen-delete.ts'; + const nodeId = `Function:${filePath}:target:1`; + + await adapter.executeQuery( + `CREATE (:Function {id: '${nodeId}', name: 'target', filePath: '${filePath}', startLine: 1, endLine: 3, isExported: true, content: '', description: ''})`, + ); + await batchInsertEmbeddings(adapter.executeWithReusedStatement, [ + { + nodeId, + chunkIndex: 0, + startLine: 1, + endLine: 3, + embedding: new Array(EMBEDDING_DIMS).fill(0), + }, + ]); + await adapter.createVectorIndex(); + await adapter.closeLbug(); + await adapter.initLbug(handle.dbPath); + + await expect(adapter.deleteNodesForFiles([filePath])).resolves.toBeUndefined(); + const rows = (await adapter.executeQuery( + `MATCH (e:CodeEmbedding) WHERE e.nodeId = '${nodeId}' RETURN count(e) AS count`, + )) as Array<{ count: number | bigint }>; + expect(Number(rows[0]?.count ?? 0)).toBe(0); + }); }); }); From 1595faccd82c8950f3e4d5a5c78b24f0c079d784 Mon Sep 17 00:00:00 2001 From: Eva Date: Tue, 14 Jul 2026 02:19:10 +0700 Subject: [PATCH 2/2] fix(embeddings): resume partial checkpoint windows --- .../src/core/embeddings/embedding-pipeline.ts | 44 +++++++-- gitnexus/src/core/run-analyze.ts | 82 ++++++++++------ gitnexus/src/server/api.ts | 65 +++++++++++-- gitnexus/src/storage/repo-manager.ts | 16 +++- .../test/unit/api-readonly-wiring.test.ts | 21 ++++ gitnexus/test/unit/embedding-pipeline.test.ts | 95 +++++++++++++++++++ gitnexus/test/unit/run-analyze.test.ts | 29 ++++++ 7 files changed, 302 insertions(+), 50 deletions(-) diff --git a/gitnexus/src/core/embeddings/embedding-pipeline.ts b/gitnexus/src/core/embeddings/embedding-pipeline.ts index b40e18e7d1..86e918cb3f 100644 --- a/gitnexus/src/core/embeddings/embedding-pipeline.ts +++ b/gitnexus/src/core/embeddings/embedding-pipeline.ts @@ -311,9 +311,15 @@ export interface EmbeddingPipelineCheckpoint { chunksProcessed: number; } +export interface EmbeddingPipelineCheckpointWindow extends EmbeddingPipelineCheckpoint { + nodeIds: string[]; +} + export interface EmbeddingPipelineOptions { signal?: AbortSignal; checkpointEveryNodes?: number; + forceReembedNodeIds?: ReadonlySet; + onCheckpointWindowStart?: (window: EmbeddingPipelineCheckpointWindow) => Promise; onCheckpoint?: (checkpoint: EmbeddingPipelineCheckpoint) => Promise; } @@ -425,6 +431,7 @@ export const runEmbeddingPipeline = async ( // Phase 2: Query embeddable nodes let nodes = await queryEmbeddableNodes(executeQuery); throwIfCancelled(); + const embeddableNodeIds = new Set(nodes.map((node) => node.id)); // Incremental mode: compare content hashes, delete stale rows, skip fresh ones. // Computed hashes for stale nodes are cached so batchInsertEmbeddings can reuse them @@ -434,16 +441,20 @@ export const runEmbeddingPipeline = async ( // than all up front — see U6 / KTD7. `staleNodeIds` is consulted inside the // batch loop; it stays empty in full (non-incremental) mode so no deletes fire. const staleNodeIds = new Set(); - if (existingEmbeddings && existingEmbeddings.size > 0) { + const forceReembedNodeIds = pipelineOptions.forceReembedNodeIds; + if ( + (existingEmbeddings && existingEmbeddings.size > 0) || + (forceReembedNodeIds && forceReembedNodeIds.size > 0) + ) { const beforeCount = nodes.length; nodes = nodes.filter((n) => { - const existingHash = existingEmbeddings.get(n.id); + const existingHash = existingEmbeddings?.get(n.id); if (existingHash === undefined) { // New node — needs embedding return true; } const currentHash = contentHashForNode(n, finalConfig); - if (currentHash !== existingHash) { + if (currentHash !== existingHash || forceReembedNodeIds?.has(n.id)) { // Content changed — cache hash for reuse during insert, mark for DELETE + re-embed computedStaleHashes.set(n.id, currentHash); staleNodeIds.add(n.id); @@ -460,6 +471,14 @@ export const runEmbeddingPipeline = async ( } } + if (forceReembedNodeIds && forceReembedNodeIds.size > 0) { + const removedPendingNodeIds = [...forceReembedNodeIds].filter( + (nodeId) => !embeddableNodeIds.has(nodeId), + ); + await deleteStaleEmbeddingRows(executeWithReusedStatement, removedPendingNodeIds); + throwIfCancelled(); + } + const totalNodes = nodes.length; if (isDev) { @@ -491,8 +510,11 @@ export const runEmbeddingPipeline = async ( const batchSize = finalConfig.batchSize; const chunkSize = finalConfig.chunkSize; const overlap = finalConfig.overlap; + const checkpointWindowNodeCount = Math.max( + batchSize, + Math.ceil(checkpointEveryNodes / batchSize) * batchSize, + ); let processedNodes = 0; - let lastCheckpointAt = 0; onProgress({ phase: 'embedding', @@ -506,6 +528,17 @@ export const runEmbeddingPipeline = async ( // Process in batches of nodes for (let batchIndex = 0; batchIndex < totalNodes; batchIndex += batchSize) { throwIfCancelled(); + if (pipelineOptions.onCheckpointWindowStart && batchIndex % checkpointWindowNodeCount === 0) { + await pipelineOptions.onCheckpointWindowStart({ + nodesProcessed: processedNodes, + totalNodes, + chunksProcessed: totalChunks, + nodeIds: nodes + .slice(batchIndex, batchIndex + checkpointWindowNodeCount) + .map((node) => node.id), + }); + throwIfCancelled(); + } const batch = nodes.slice(batchIndex, batchIndex + batchSize); // Chunk each node and generate text @@ -631,14 +664,13 @@ export const runEmbeddingPipeline = async ( if ( pipelineOptions.onCheckpoint && - (processedNodes - lastCheckpointAt >= checkpointEveryNodes || processedNodes === totalNodes) + (processedNodes % checkpointWindowNodeCount === 0 || processedNodes === totalNodes) ) { await pipelineOptions.onCheckpoint({ nodesProcessed: processedNodes, totalNodes, chunksProcessed: totalChunks, }); - lastCheckpointAt = processedNodes; throwIfCancelled(); } } diff --git a/gitnexus/src/core/run-analyze.ts b/gitnexus/src/core/run-analyze.ts index debfc073e1..f8622a4bc1 100644 --- a/gitnexus/src/core/run-analyze.ts +++ b/gitnexus/src/core/run-analyze.ts @@ -856,6 +856,7 @@ export async function runFullAnalysis( } let resumeEmbeddingCheckpoint = false; + let pendingEmbeddingNodeIds = new Set(); let embeddingIdentityForRun: EmbeddingIdentity | undefined; if (existingMeta?.embeddingCheckpoint) { if (options.dropEmbeddings) { @@ -876,9 +877,11 @@ export async function runFullAnalysis( ); } resumeEmbeddingCheckpoint = true; + pendingEmbeddingNodeIds = new Set(checkpoint.pendingNodeIds ?? []); log( `Previous analyze ended at an embedding checkpoint ` + - `(${checkpoint.nodesProcessed}/${checkpoint.totalNodes} nodes); resuming from persisted hashes.`, + `(${checkpoint.nodesProcessed}/${checkpoint.totalNodes} nodes); resuming from persisted hashes` + + `${pendingEmbeddingNodeIds.size > 0 ? ` and regenerating ${pendingEmbeddingNodeIds.size} pending node(s)` : ''}.`, ); } } @@ -2005,6 +2008,48 @@ export async function runFullAnalysis( } } + const saveEmbeddingCheckpoint = async ( + checkpoint: { + nodesProcessed: number; + totalNodes: number; + chunksProcessed: number; + }, + pendingNodeIds: string[], + embeddings: number | undefined, + ): Promise => { + const fileHashes: Record = {}; + for (const [key, value] of newFileHashes) fileHashes[key] = value; + await saveMeta(metaDir, { + ...(existingMeta ?? {}), + repoPath, + lastCommit: currentCommit, + indexedAt: new Date().toISOString(), + branch: branchLabel ?? existingMeta?.branch, + remoteUrl: hasGitDir(repoPath) ? getRemoteUrl(repoPath) : undefined, + stats: { + files: pipelineResult.totalFileCount, + nodes: stats.nodes, + edges: stats.edges, + communities: pipelineResult.communityResult?.stats.totalCommunities, + processes: pipelineResult.processResult?.stats.totalProcesses, + embeddings, + }, + schemaVersion: hasGitDir(repoPath) ? INCREMENTAL_SCHEMA_VERSION : undefined, + cjkSegmentation: getSearchFTSCjkSegmentation(), + fileHashes: hasGitDir(repoPath) ? fileHashes : undefined, + cacheKeys: [...parseCache.usedKeys], + incrementalInProgress: undefined, + embeddingCheckpoint: { + at: new Date().toISOString(), + ...checkpoint, + model: embeddingIdentity.model, + dimensions: embeddingIdentity.dimensions, + pendingNodeIds, + }, + pdg: resolvePdgConfig(options), + }); + }; + const embeddingResult = await runEmbeddingPipeline( executeQuery, executeWithReusedStatement, @@ -2022,6 +2067,10 @@ export async function runFullAnalysis( cachedEmbeddingNodeIds.size > 0 ? cachedEmbeddingNodeIds : undefined, existingEmbeddings, { + forceReembedNodeIds: pendingEmbeddingNodeIds, + onCheckpointWindowStart: async ({ nodeIds, ...checkpoint }) => { + await saveEmbeddingCheckpoint(checkpoint, nodeIds, existingMeta?.stats?.embeddings); + }, onCheckpoint: async (checkpoint) => { await checkpointOnce(); const countResult = await executeQuery( @@ -2029,36 +2078,7 @@ export async function runFullAnalysis( ); const countRow = countResult?.[0]; const embeddings = Number(countRow?.cnt ?? countRow?.[0] ?? 0); - const fileHashes: Record = {}; - for (const [key, value] of newFileHashes) fileHashes[key] = value; - await saveMeta(metaDir, { - ...(existingMeta ?? {}), - repoPath, - lastCommit: currentCommit, - indexedAt: new Date().toISOString(), - branch: branchLabel ?? existingMeta?.branch, - remoteUrl: hasGitDir(repoPath) ? getRemoteUrl(repoPath) : undefined, - stats: { - files: pipelineResult.totalFileCount, - nodes: stats.nodes, - edges: stats.edges, - communities: pipelineResult.communityResult?.stats.totalCommunities, - processes: pipelineResult.processResult?.stats.totalProcesses, - embeddings, - }, - schemaVersion: hasGitDir(repoPath) ? INCREMENTAL_SCHEMA_VERSION : undefined, - cjkSegmentation: getSearchFTSCjkSegmentation(), - fileHashes: hasGitDir(repoPath) ? fileHashes : undefined, - cacheKeys: [...parseCache.usedKeys], - incrementalInProgress: undefined, - embeddingCheckpoint: { - at: new Date().toISOString(), - ...checkpoint, - model: embeddingIdentity.model, - dimensions: embeddingIdentity.dimensions, - }, - pdg: resolvePdgConfig(options), - }); + await saveEmbeddingCheckpoint(checkpoint, [], embeddings); }, }, ); diff --git a/gitnexus/src/server/api.ts b/gitnexus/src/server/api.ts index b7e2930cb6..50b461f48d 100644 --- a/gitnexus/src/server/api.ts +++ b/gitnexus/src/server/api.ts @@ -17,6 +17,7 @@ import { canonicalizePath, cloneDirBelongsToEntry, loadMeta, + saveMeta, listRegisteredRepos, getStoragePath, registryPathEquals, @@ -1784,7 +1785,6 @@ export const createServer = async (port: number, host: string = '127.0.0.1') => const embedTimeout = setTimeout(() => { const current = embedJobManager.getJob(job.id); if (current && current.status !== 'complete' && current.status !== 'failed') { - releaseRepoLock(repoLockPath); embedJobManager.cancelJob(job.id, 'Embedding timed out (30 minute limit)'); } }, EMBED_TIMEOUT_MS); @@ -1796,6 +1796,50 @@ 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'); + const [{ getEmbeddingDimensions }, { resolveEmbeddingConfig }] = await Promise.all([ + import('../core/embeddings/embedder.js'), + import('../core/embeddings/config.js'), + ]); + const embeddingIdentity = { + model: process.env.GITNEXUS_EMBEDDING_MODEL ?? resolveEmbeddingConfig().modelId, + dimensions: getEmbeddingDimensions(), + }; + let embeddingMeta = await loadMeta(entry.storagePath); + if (!embeddingMeta) { + throw new Error('Repository metadata is missing; run gitnexus analyze first'); + } + const priorCheckpoint = embeddingMeta.embeddingCheckpoint; + if ( + priorCheckpoint && + (priorCheckpoint.model !== embeddingIdentity.model || + priorCheckpoint.dimensions !== embeddingIdentity.dimensions) + ) { + throw new Error( + `Cannot resume embedding checkpoint: it uses ${priorCheckpoint.model} at ` + + `${priorCheckpoint.dimensions} dimensions, but this run resolves ` + + `${embeddingIdentity.model} at ${embeddingIdentity.dimensions}.`, + ); + } + const forceReembedNodeIds = new Set(priorCheckpoint?.pendingNodeIds ?? []); + const saveEmbeddingCheckpoint = async ( + checkpoint: { + nodesProcessed: number; + totalNodes: number; + chunksProcessed: number; + }, + pendingNodeIds: string[], + ): Promise => { + embeddingMeta = { + ...embeddingMeta, + embeddingCheckpoint: { + at: new Date().toISOString(), + ...checkpoint, + ...embeddingIdentity, + pendingNodeIds, + }, + }; + await saveMeta(entry.storagePath, embeddingMeta); + }; // 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'); @@ -1832,11 +1876,13 @@ export const createServer = async (port: number, host: string = '127.0.0.1') => existingEmbeddings, { signal: embedController.signal, - onCheckpoint: async () => { - // Make each reported pipeline checkpoint durable before the - // next batch starts. A cancelled/restarted job then resumes - // from content hashes already present in LadybugDB. + forceReembedNodeIds, + onCheckpointWindowStart: async ({ nodeIds, ...checkpoint }) => { + await saveEmbeddingCheckpoint(checkpoint, nodeIds); + }, + onCheckpoint: async (checkpoint) => { await flushWAL(); + await saveEmbeddingCheckpoint(checkpoint, []); }, }, ); @@ -1846,18 +1892,16 @@ export const createServer = async (port: number, host: string = '127.0.0.1') => // handles this during process exit, but the server keeps the // connection open for other routes — a CHECKPOINT is enough. await flushWAL(); + embeddingMeta = { ...embeddingMeta, embeddingCheckpoint: undefined }; + await saveMeta(entry.storagePath, embeddingMeta); }); - clearTimeout(embedTimeout); - releaseRepoLock(repoLockPath); // Don't overwrite 'failed' if the job was cancelled while the pipeline was running const current = embedJobManager.getJob(job.id); if (!current || current.status !== 'failed') { embedJobManager.updateJob(job.id, { status: 'complete' }); } } catch (err: any) { - clearTimeout(embedTimeout); - releaseRepoLock(repoLockPath); const current = embedJobManager.getJob(job.id); if (!current || current.status !== 'failed') { embedJobManager.updateJob(job.id, { @@ -1865,6 +1909,9 @@ export const createServer = async (port: number, host: string = '127.0.0.1') => error: err.message || 'Embedding generation failed', }); } + } finally { + clearTimeout(embedTimeout); + releaseRepoLock(repoLockPath); } })(); diff --git a/gitnexus/src/storage/repo-manager.ts b/gitnexus/src/storage/repo-manager.ts index 6839eebc4e..942f45ddb2 100644 --- a/gitnexus/src/storage/repo-manager.ts +++ b/gitnexus/src/storage/repo-manager.ts @@ -200,10 +200,11 @@ export interface RepoMeta { droppedImporterChunks?: number; }; /** - * Durable marker written after a completed embedding batch and LadybugDB - * checkpoint. The graph is valid, but embedding generation has not reached - * final metadata registration. A matching runtime resumes from persisted - * content hashes; a model or dimension mismatch fails before mutation. + * Durable embedding-resume marker. Before a bounded write window begins, + * `pendingNodeIds` records every node that could become partially persisted; + * after the LadybugDB checkpoint it is cleared while progress is retained. + * A matching runtime resumes from persisted hashes and regenerates pending + * nodes; a model or dimension mismatch fails before mutation. */ embeddingCheckpoint?: { at: string; @@ -212,6 +213,13 @@ export interface RepoMeta { chunksProcessed: number; model: string; dimensions: number; + /** + * Nodes in the current checkpoint window. Any of these may have only a + * subset of their chunks persisted after an abrupt process termination, + * so resume must delete and regenerate them even when a persisted row has + * the current content hash. + */ + pendingNodeIds?: string[]; }; /** * Name of the git branch this index represents (#2106). Absent for the diff --git a/gitnexus/test/unit/api-readonly-wiring.test.ts b/gitnexus/test/unit/api-readonly-wiring.test.ts index 7bb8681dc6..8e86e7f31d 100644 --- a/gitnexus/test/unit/api-readonly-wiring.test.ts +++ b/gitnexus/test/unit/api-readonly-wiring.test.ts @@ -57,4 +57,25 @@ describe('api read-only endpoint wiring', () => { expect(embedSection[0]).not.toMatch(/readOnly:\s*true/); } }); + + it('/api/embed keeps the repository lock until cancelled work actually stops', async () => { + const source = await readSource(); + const timeoutSection = source.match( + /const embedTimeout = setTimeout\([\s\S]*?\/\/ Run embedding pipeline asynchronously/, + ); + expect(timeoutSection).not.toBeNull(); + expect(timeoutSection?.[0]).not.toContain('releaseRepoLock(repoLockPath)'); + }); + + it('/api/embed persists and resumes bounded pending windows', async () => { + const source = await readSource(); + const embedSection = source.match( + /\/\/ Run embedding pipeline asynchronously[\s\S]*?res\.status\(202\)/, + ); + expect(embedSection).not.toBeNull(); + expect(embedSection?.[0]).toContain('forceReembedNodeIds'); + expect(embedSection?.[0]).toContain('onCheckpointWindowStart'); + expect(embedSection?.[0]).toContain('pendingNodeIds'); + expect(embedSection?.[0]).toContain('saveMeta'); + }); }); diff --git a/gitnexus/test/unit/embedding-pipeline.test.ts b/gitnexus/test/unit/embedding-pipeline.test.ts index 68ba7fd4b2..8a874e47eb 100644 --- a/gitnexus/test/unit/embedding-pipeline.test.ts +++ b/gitnexus/test/unit/embedding-pipeline.test.ts @@ -709,6 +709,101 @@ describe('runEmbeddingPipeline incremental filter', () => { expect(resumedIds).toEqual([second.id]); }); + it('re-embeds a pending-window node even when its persisted content hash matches', async () => { + mockEmbedderSetup(); + const node = makeNode({ + id: 'Function:pending:src/pending.ts', + name: 'pending', + filePath: 'src/pending.ts', + }); + const currentHash = contentHashForNode(node, DEFAULT_EMBEDDING_CONFIG); + const executeQuery = mockExecuteQuery([node]); + const executeWithReusedStatement = mockExecuteWithReusedStatement(); + const { runEmbeddingPipeline } = + await import('../../src/core/embeddings/embedding-pipeline.js'); + + await runEmbeddingPipeline( + executeQuery, + executeWithReusedStatement, + onProgress, + {}, + undefined, + new Map([[node.id, currentHash]]), + { forceReembedNodeIds: new Set([node.id]) }, + ); + + const deletedIds = stmtCalls + .filter((call) => call.cypher.includes('DELETE')) + .flatMap((call) => call.params.map((param) => param.nodeId)); + const insertedIds = stmtCalls + .filter((call) => call.cypher.includes('CREATE')) + .flatMap((call) => call.params.map((param) => param.nodeId)); + expect(deletedIds).toContain(node.id); + expect(insertedIds).toContain(node.id); + }); + + it('announces each checkpoint window before mutating any node in that window', async () => { + mockEmbedderSetup(); + const first = makeNode({ id: 'Function:first:src/first.ts', name: 'first' }); + const second = makeNode({ id: 'Function:second:src/second.ts', name: 'second' }); + const third = makeNode({ id: 'Function:third:src/third.ts', name: 'third' }); + const executeQuery = mockExecuteQuery([first, second, third]); + const executeWithReusedStatement = mockExecuteWithReusedStatement(); + const windows: string[][] = []; + const mutationCountsAtWindowStart: number[] = []; + const checkpoints: number[] = []; + const { runEmbeddingPipeline } = + await import('../../src/core/embeddings/embedding-pipeline.js'); + + await runEmbeddingPipeline( + executeQuery, + executeWithReusedStatement, + onProgress, + { batchSize: 1 }, + undefined, + new Map(), + { + checkpointEveryNodes: 2, + onCheckpointWindowStart: async ({ nodeIds }) => { + windows.push(nodeIds); + mutationCountsAtWindowStart.push(stmtCalls.length); + }, + onCheckpoint: async ({ nodesProcessed }) => { + checkpoints.push(nodesProcessed); + }, + }, + ); + + expect(windows).toEqual([[first.id, second.id], [third.id]]); + expect(mutationCountsAtWindowStart).toEqual([0, 2]); + expect(checkpoints).toEqual([2, 3]); + }); + + it('deletes pending-window rows whose node is no longer embeddable', async () => { + mockEmbedderSetup(); + const live = makeNode({ id: 'Function:live:src/live.ts', name: 'live' }); + const removedNodeId = 'Function:removed:src/removed.ts'; + const executeQuery = mockExecuteQuery([live]); + const executeWithReusedStatement = mockExecuteWithReusedStatement(); + const { runEmbeddingPipeline } = + await import('../../src/core/embeddings/embedding-pipeline.js'); + + await runEmbeddingPipeline( + executeQuery, + executeWithReusedStatement, + onProgress, + {}, + undefined, + new Map([[removedNodeId, 'persisted-partial-hash']]), + { forceReembedNodeIds: new Set([removedNodeId]) }, + ); + + const deletedIds = stmtCalls + .filter((call) => call.cypher.includes('DELETE')) + .flatMap((call) => call.params.map((param) => param.nodeId)); + expect(deletedIds).toContain(removedNodeId); + }); + it('deletes only stale nodes — new and unchanged nodes are never deleted (#2333 U6)', async () => { mockEmbedderSetup(); diff --git a/gitnexus/test/unit/run-analyze.test.ts b/gitnexus/test/unit/run-analyze.test.ts index b95f6a00fd..af2c64ad80 100644 --- a/gitnexus/test/unit/run-analyze.test.ts +++ b/gitnexus/test/unit/run-analyze.test.ts @@ -17,6 +17,7 @@ import { } from '../../src/storage/repo-manager.js'; import { taintModelVersion } from '../../src/core/ingestion/taint/typescript-model.js'; import { createTempDir } from '../helpers/test-db.js'; +import { readEmbeddingNodeIds } from '../helpers/embedding-seed.js'; describe('run-analyze module', () => { it('exports runFullAnalysis as a function', async () => { @@ -150,8 +151,36 @@ describe('run-analyze module', () => { const finalized = await loadMeta(storagePath); if (!finalized) throw new Error('expected finalized metadata'); + const [pendingNodeId] = await readEmbeddingNodeIds(tmpRepo.dbPath); + if (!pendingNodeId) throw new Error('expected a persisted embedding node'); await saveMeta(storagePath, { ...finalized, + embeddingCheckpoint: { + at: new Date().toISOString(), + nodesProcessed: 0, + totalNodes: 1, + chunksProcessed: 0, + model: 'test-model', + dimensions: 384, + pendingNodeIds: [pendingNodeId], + }, + }); + fetchMock.mockClear(); + + await runFullAnalysis( + tmpRepo.dbPath, + { skipAgentsMd: true, skipSkills: true }, + { onProgress: () => {} }, + ); + + expect(fetchMock).toHaveBeenCalled(); + expect((await loadMeta(storagePath))?.embeddingCheckpoint).toBeUndefined(); + + const resumedPending = await loadMeta(storagePath); + if (!resumedPending) throw new Error('expected pending-window resume metadata'); + fetchMock.mockClear(); + await saveMeta(storagePath, { + ...resumedPending, embeddingCheckpoint: { at: new Date().toISOString(), nodesProcessed: 1,