From 554c5181cf6853325347e56545ee406949059626 Mon Sep 17 00:00:00 2001 From: Eva Date: Tue, 14 Jul 2026 01:40:47 +0700 Subject: [PATCH 1/9] fix(embeddings): fall back to text-bearing file nodes --- .../src/core/embeddings/embedding-pipeline.ts | 45 +++++++++++- gitnexus/test/unit/embedding-pipeline.test.ts | 72 +++++++++++++++++++ 2 files changed, 116 insertions(+), 1 deletion(-) diff --git a/gitnexus/src/core/embeddings/embedding-pipeline.ts b/gitnexus/src/core/embeddings/embedding-pipeline.ts index 7b09e7314..d1fa90034 100644 --- a/gitnexus/src/core/embeddings/embedding-pipeline.ts +++ b/gitnexus/src/core/embeddings/embedding-pipeline.ts @@ -180,7 +180,50 @@ const queryEmbeddableNodes = async ( } } - return allNodes; + return allNodes.length > 0 ? allNodes : queryFallbackFileNodes(executeQuery); +}; + +/** + * Static and documentation repositories may contain no code symbols while + * still persisting useful text on File nodes. Keep File embeddings as a + * zero-symbol fallback so code repositories retain symbol-first selection. + */ +const queryFallbackFileNodes = async ( + executeQuery: (cypher: string) => Promise, +): Promise => { + try { + const rows = await executeQuery(` + MATCH (n:File) + RETURN n.id AS id, n.name AS name, 'File' AS label, + n.filePath AS filePath, n.content AS content + `); + + return rows + .map((row) => { + const content = row.content ?? row[4] ?? ''; + return { + id: row.id ?? row[0], + name: row.name ?? row[1], + label: row.label ?? row[2] ?? 'File', + filePath: row.filePath ?? row[3], + content, + startLine: 1, + endLine: Math.max(1, content.split('\n').length), + }; + }) + .filter( + (node) => + node.id && + node.filePath && + node.content.trim() && + node.content !== '[Binary file - content not stored]', + ); + } catch (error) { + if (isDev) { + logger.warn({ error }, 'Fallback File-node embedding query failed:'); + } + return []; + } }; /** diff --git a/gitnexus/test/unit/embedding-pipeline.test.ts b/gitnexus/test/unit/embedding-pipeline.test.ts index 570593a2d..50443c724 100644 --- a/gitnexus/test/unit/embedding-pipeline.test.ts +++ b/gitnexus/test/unit/embedding-pipeline.test.ts @@ -289,6 +289,78 @@ describe('runEmbeddingPipeline incremental filter', () => { progressUpdates.push({ ...p }); }; + it('falls back to text-bearing File nodes when a repo has no code symbols', async () => { + mockEmbedderSetup(); + + const fileNode = makeNode({ + id: 'File:README.md', + name: 'README.md', + label: 'File', + filePath: 'README.md', + content: '# Static Site\n\nDeployment and recovery notes.', + startLine: 1, + endLine: 3, + }); + const emptyFile = makeNode({ + id: 'File:empty.txt', + name: 'empty.txt', + label: 'File', + filePath: 'empty.txt', + content: ' ', + }); + const binaryFile = makeNode({ + id: 'File:logo.png', + name: 'logo.png', + label: 'File', + filePath: 'logo.png', + content: '[Binary file - content not stored]', + }); + const executeQuery = mockExecuteQuery([fileNode, emptyFile, binaryFile]); + const executeWithReusedStatement = mockExecuteWithReusedStatement(); + + const { runEmbeddingPipeline } = + await import('../../src/core/embeddings/embedding-pipeline.js'); + + const result = await runEmbeddingPipeline(executeQuery, executeWithReusedStatement, onProgress); + + expect(queryCalls.some((cypher) => cypher.includes('MATCH (n:File)'))).toBe(true); + const insertedNodeIds = stmtCalls + .filter((call) => call.cypher.includes('CREATE')) + .flatMap((call) => call.params.map((param) => param.nodeId)); + expect(insertedNodeIds).toContain(fileNode.id); + expect(insertedNodeIds).not.toContain(emptyFile.id); + expect(insertedNodeIds).not.toContain(binaryFile.id); + expect(result.nodesProcessed).toBe(1); + }); + + it('retains symbol-first selection when code symbols exist', async () => { + mockEmbedderSetup(); + + const functionNode = makeNode(); + const fileNode = makeNode({ + id: 'File:src/main.ts', + name: 'main.ts', + label: 'File', + filePath: 'src/main.ts', + content: 'function foo() { return 1; }', + }); + const executeQuery = mockExecuteQuery([functionNode, fileNode]); + const executeWithReusedStatement = mockExecuteWithReusedStatement(); + + const { runEmbeddingPipeline } = + await import('../../src/core/embeddings/embedding-pipeline.js'); + + const result = await runEmbeddingPipeline(executeQuery, executeWithReusedStatement, onProgress); + + expect(queryCalls.some((cypher) => cypher.includes('MATCH (n:File)'))).toBe(false); + const insertedNodeIds = stmtCalls + .filter((call) => call.cypher.includes('CREATE')) + .flatMap((call) => call.params.map((param) => param.nodeId)); + expect(insertedNodeIds).toContain(functionNode.id); + expect(insertedNodeIds).not.toContain(fileNode.id); + expect(result.nodesProcessed).toBe(1); + }); + it('skips unchanged nodes when hash matches', async () => { mockEmbedderSetup(); From 711ff8721dbf13f9f28dcf5421189b7fb1bca196 Mon Sep 17 00:00:00 2001 From: Eva Date: Tue, 14 Jul 2026 02:15:57 +0700 Subject: [PATCH 2/9] fix(embeddings): make HTTP generation resumable --- .../src/integrations/resilient-fetch.ts | 8 +- gitnexus/.env.example | 3 + gitnexus/README.md | 5 +- gitnexus/src/core/embeddings/embedder.ts | 24 ++- .../src/core/embeddings/embedding-pipeline.ts | 81 +++++++- gitnexus/src/core/embeddings/http-client.ts | 153 ++++++++++++++- gitnexus/src/core/run-analyze.ts | 122 +++++++++++- gitnexus/src/server/analyze-job.ts | 17 ++ gitnexus/src/server/api.ts | 73 ++++++- gitnexus/src/storage/repo-manager.ts | 22 +++ gitnexus/test/unit/analyze-job.test.ts | 11 ++ .../test/unit/api-readonly-wiring.test.ts | 21 ++ gitnexus/test/unit/embedding-pipeline.test.ts | 183 ++++++++++++++++++ gitnexus/test/unit/http-embedder.test.ts | 128 ++++++++++++ .../unit/integrations/resilient-fetch.test.ts | 16 ++ gitnexus/test/unit/run-analyze.test.ts | 142 +++++++++++++- 16 files changed, 975 insertions(+), 34 deletions(-) diff --git a/gitnexus-shared/src/integrations/resilient-fetch.ts b/gitnexus-shared/src/integrations/resilient-fetch.ts index c91b9db3a..828dea731 100644 --- a/gitnexus-shared/src/integrations/resilient-fetch.ts +++ b/gitnexus-shared/src/integrations/resilient-fetch.ts @@ -33,6 +33,8 @@ export interface ResilientFetchOptions { breakerOptions?: CircuitBreakerOptions; /** Tuning knobs for the retry helper. */ retry?: Partial> & { + /** Upper bound on a single Retry-After wait. Defaults to RETRY_AFTER_CAP_MS. */ + retryAfterCapMs?: number; sleep?: RetryOptions['sleep']; random?: RetryOptions['random']; }; @@ -83,6 +85,7 @@ type Outcome = export function classifyOutcome( result: { kind: 'error'; err: unknown } | { kind: 'response'; resp: Response }, now: () => number, + retryAfterCapMs = RETRY_AFTER_CAP_MS, ): Outcome { if (result.kind === 'error') { // Both timer-fired aborts (`AbortSignal.timeout()` → `TimeoutError`) @@ -111,7 +114,7 @@ export function classifyOutcome( return { kind: 'retryable-status', resp, - afterMs: parsed !== null ? Math.min(parsed, RETRY_AFTER_CAP_MS) : undefined, + afterMs: parsed !== null ? Math.min(parsed, retryAfterCapMs) : undefined, }; } if (resp.status >= 500) return { kind: 'retryable-status', resp, afterMs: undefined }; @@ -176,6 +179,7 @@ export async function resilientFetch( maxAttempts: opts.retry?.maxAttempts ?? DEFAULT_RETRY.maxAttempts, baseDelayMs: opts.retry?.baseDelayMs ?? DEFAULT_RETRY.baseDelayMs, capDelayMs: opts.retry?.capDelayMs ?? DEFAULT_RETRY.capDelayMs, + retryAfterCapMs: opts.retry?.retryAfterCapMs ?? RETRY_AFTER_CAP_MS, }; const sleep = opts.retry?.sleep ?? defaultSleep; const random = opts.retry?.random ?? Math.random; @@ -202,7 +206,7 @@ export async function resilientFetch( result = { kind: 'error', err }; } - const outcome = classifyOutcome(result, now); + const outcome = classifyOutcome(result, now, retryConfig.retryAfterCapMs); switch (outcome.kind) { case 'success': diff --git a/gitnexus/.env.example b/gitnexus/.env.example index 8f2f83dc4..8b90c7ae7 100644 --- a/gitnexus/.env.example +++ b/gitnexus/.env.example @@ -7,6 +7,9 @@ # GITNEXUS_EMBEDDING_MODEL=BAAI/bge-large-en-v1.5 # GITNEXUS_EMBEDDING_DIMS=1024 # GITNEXUS_EMBEDDING_API_KEY=your-key +# GITNEXUS_EMBEDDING_MAX_ATTEMPTS=3 +# GITNEXUS_EMBEDDING_RETRY_CAP_MS=5000 +# GITNEXUS_EMBEDDING_MIN_INTERVAL_MS=0 # Works with Infinity, vLLM, TEI, llama.cpp, Ollama, LM Studio, or OpenAI. # See README for details. diff --git a/gitnexus/README.md b/gitnexus/README.md index 2dd0d8180..6ca03acfc 100644 --- a/gitnexus/README.md +++ b/gitnexus/README.md @@ -283,10 +283,13 @@ export GITNEXUS_EMBEDDING_URL=http://your-server:8080/v1 export GITNEXUS_EMBEDDING_MODEL=BAAI/bge-large-en-v1.5 export GITNEXUS_EMBEDDING_DIMS=1024 # optional, default 384 export GITNEXUS_EMBEDDING_API_KEY=your-key # optional, default: "unused" +export GITNEXUS_EMBEDDING_MAX_ATTEMPTS=3 # optional, total attempts (1-20) +export GITNEXUS_EMBEDDING_RETRY_CAP_MS=5000 # optional, maximum retry delay +export GITNEXUS_EMBEDDING_MIN_INTERVAL_MS=0 # optional, minimum request spacing gitnexus analyze . --embeddings ``` -Works with Infinity, vLLM, TEI, llama.cpp, Ollama, LM Studio, or OpenAI. When unset, local embeddings are used unchanged. +Works with Infinity, vLLM, TEI, llama.cpp, Ollama, LM Studio, or OpenAI. Retry and pacing settings are provider-neutral; provider-specific limits should be supplied through configuration. When unset, local embeddings are used unchanged. ## Multi-Repo Support diff --git a/gitnexus/src/core/embeddings/embedder.ts b/gitnexus/src/core/embeddings/embedder.ts index 9c4594a4d..e41c4f4c9 100644 --- a/gitnexus/src/core/embeddings/embedder.ts +++ b/gitnexus/src/core/embeddings/embedder.ts @@ -20,7 +20,12 @@ if (!process.env.ORT_LOG_LEVEL) { // initEmbedder, after the platform guard has passed (#1515). import type { FeatureExtractionPipeline, ProgressInfo } from '@huggingface/transformers'; import { DEFAULT_EMBEDDING_CONFIG, type EmbeddingConfig, type ModelProgress } from './types.js'; -import { isHttpMode, getHttpDimensions, httpEmbed } from './http-client.js'; +import { + isHttpMode, + getHttpDimensions, + httpEmbed, + type EmbeddingRequestOptions, +} from './http-client.js'; import { resolveEmbeddingConfig } from './config.js'; import { applyHfEnvOverrides, isHfDownloadFailure, withHfDownloadRetry } from './hf-env.js'; import { @@ -297,9 +302,13 @@ export const getEmbedder = (): FeatureExtractionPipeline => { * @param text - Text to embed * @returns Float32Array of embedding vector */ -export const embedText = async (text: string): Promise => { +export const embedText = async ( + text: string, + options: EmbeddingRequestOptions = {}, +): Promise => { + options.signal?.throwIfAborted(); if (isHttpMode()) { - const [vec] = await httpEmbed([text]); + const [vec] = await httpEmbed([text], options); return vec; } @@ -321,13 +330,17 @@ export const embedText = async (text: string): Promise => { * @param texts - Array of texts to embed * @returns Array of Float32Array embedding vectors */ -export const embedBatch = async (texts: string[]): Promise => { +export const embedBatch = async ( + texts: string[], + options: EmbeddingRequestOptions = {}, +): Promise => { + options.signal?.throwIfAborted(); if (texts.length === 0) { return []; } if (isHttpMode()) { - return httpEmbed(texts); + return httpEmbed(texts, options); } const embedder = getEmbedder(); @@ -337,6 +350,7 @@ export const embedBatch = async (texts: string[]): Promise => { pooling: 'mean', normalize: true, }); + options.signal?.throwIfAborted(); // Result shape is [batch_size, dimensions] // Need to split into individual vectors diff --git a/gitnexus/src/core/embeddings/embedding-pipeline.ts b/gitnexus/src/core/embeddings/embedding-pipeline.ts index 7b09e7314..37580b6f5 100644 --- a/gitnexus/src/core/embeddings/embedding-pipeline.ts +++ b/gitnexus/src/core/embeddings/embedding-pipeline.ts @@ -262,6 +262,24 @@ export interface EmbeddingPipelineResult { semanticMode: 'vector-index' | 'exact-scan'; } +export interface EmbeddingPipelineCheckpoint { + nodesProcessed: number; + totalNodes: number; + 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; +} + /** * DELETE stale embedding rows for the given nodeIds so they can be re-inserted. * @@ -320,12 +338,20 @@ export const runEmbeddingPipeline = async ( config: Partial = {}, skipNodeIds?: Set, existingEmbeddings?: Map, + pipelineOptions: EmbeddingPipelineOptions = {}, ): Promise => { const finalConfig = resolveEmbeddingConfig(config); let totalChunks = 0; + const checkpointEveryNodes = pipelineOptions.checkpointEveryNodes ?? 5_000; + if (!Number.isSafeInteger(checkpointEveryNodes) || checkpointEveryNodes <= 0) { + throw new Error('checkpointEveryNodes must be a positive integer'); + } + const throwIfCancelled = (): void => pipelineOptions.signal?.throwIfAborted(); try { + throwIfCancelled(); const vectorAvailable = await ensureVectorExtensionAvailable(); + throwIfCancelled(); if (!vectorAvailable) { logger.warn(vectorUnavailableMessage); } @@ -346,6 +372,7 @@ export const runEmbeddingPipeline = async ( modelDownloadPercent: downloadPercent, }); }, finalConfig); + throwIfCancelled(); } onProgress({ @@ -360,6 +387,8 @@ 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 @@ -369,16 +398,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); @@ -395,6 +428,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) { @@ -402,6 +443,7 @@ export const runEmbeddingPipeline = async ( } if (totalNodes === 0) { + throwIfCancelled(); // 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. @@ -425,6 +467,10 @@ 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; onProgress({ @@ -438,6 +484,18 @@ 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 @@ -520,6 +578,7 @@ export const runEmbeddingPipeline = async ( // Preserves Kuzu's required DELETE-before-INSERT for vector-indexed rows. const batchStaleIds = batch.filter((n) => staleNodeIds.has(n.id)).map((n) => n.id); await deleteStaleEmbeddingRows(executeWithReusedStatement, batchStaleIds); + throwIfCancelled(); // Embed chunk texts in sub-batches to control memory const EMBED_SUB_BATCH = finalConfig.subBatchSize; @@ -529,7 +588,7 @@ export const runEmbeddingPipeline = async ( let embeddings: Float32Array[]; try { - embeddings = await embedBatch(subTexts); + embeddings = await embedBatch(subTexts, { signal: pipelineOptions.signal }); } catch (embedErr) { logger.error( { embedErr }, @@ -544,6 +603,7 @@ export const runEmbeddingPipeline = async ( })); await batchInsertEmbeddings(executeWithReusedStatement, dbUpdates); + throwIfCancelled(); } processedNodes += batch.length; @@ -558,9 +618,22 @@ export const runEmbeddingPipeline = async ( currentBatch: Math.floor(batchIndex / batchSize) + 1, totalBatches: Math.ceil(totalNodes / batchSize), }); + + if ( + pipelineOptions.onCheckpoint && + (processedNodes % checkpointWindowNodeCount === 0 || processedNodes === totalNodes) + ) { + await pipelineOptions.onCheckpoint({ + nodesProcessed: processedNodes, + totalNodes, + chunksProcessed: totalChunks, + }); + throwIfCancelled(); + } } // Phase 4: Create vector index + throwIfCancelled(); onProgress({ phase: 'indexing', percent: 90, diff --git a/gitnexus/src/core/embeddings/http-client.ts b/gitnexus/src/core/embeddings/http-client.ts index 121790251..c85d9ff02 100644 --- a/gitnexus/src/core/embeddings/http-client.ts +++ b/gitnexus/src/core/embeddings/http-client.ts @@ -16,6 +16,7 @@ import { CircuitOpenError, ResilientFetchExhaustedError, resilientFetch } from ' const HTTP_TIMEOUT_MS = 30_000; const HTTP_MAX_RETRIES = 2; const HTTP_RETRY_BACKOFF_MS = 1_000; +const HTTP_RETRY_CAP_MS = 5_000; const HTTP_BATCH_SIZE = 64; const DEFAULT_DIMS = 384; const HTTP_BREAKER_KEY = 'embeddings-http'; @@ -25,8 +26,85 @@ interface HttpConfig { model: string; apiKey: string; dimensions?: number; + maxAttempts: number; + retryCapMs: number; + minIntervalMs: number; } +export interface EmbeddingRequestOptions { + signal?: AbortSignal; +} + +let lastHttpRequestStartedAt: number | undefined; +let httpPaceQueue: Promise = Promise.resolve(); + +const parsePositiveIntegerEnv = (name: string, fallback: number, max: number): number => { + const raw = process.env[name]; + if (raw === undefined || raw === '') return fallback; + if (!/^\d+$/u.test(raw)) { + throw new Error(`${name} must be a positive integer, got "${raw}"`); + } + const parsed = Number(raw); + if (!Number.isSafeInteger(parsed) || parsed <= 0 || parsed > max) { + throw new Error(`${name} must be a positive integer <= ${max}, got "${raw}"`); + } + return parsed; +}; + +const parseNonNegativeIntegerEnv = (name: string, fallback: number, max: number): number => { + const raw = process.env[name]; + if (raw === undefined || raw === '') return fallback; + if (!/^\d+$/u.test(raw)) { + throw new Error(`${name} must be a non-negative integer, got "${raw}"`); + } + const parsed = Number(raw); + if (!Number.isSafeInteger(parsed) || parsed < 0 || parsed > max) { + throw new Error(`${name} must be a non-negative integer <= ${max}, got "${raw}"`); + } + return parsed; +}; + +const cancelledError = (): DOMException => + new DOMException('Embedding request cancelled', 'AbortError'); + +const throwIfAborted = (signal?: AbortSignal): void => { + if (signal?.aborted) throw cancelledError(); +}; + +const abortableSleep = (ms: number, signal?: AbortSignal): Promise => { + throwIfAborted(signal); + if (ms <= 0) return Promise.resolve(); + return new Promise((resolve, reject) => { + const timer = setTimeout(() => { + signal?.removeEventListener('abort', onAbort); + resolve(); + }, ms); + const onAbort = () => { + clearTimeout(timer); + signal?.removeEventListener('abort', onAbort); + reject(cancelledError()); + }; + signal?.addEventListener('abort', onAbort, { once: true }); + }); +}; + +const paceHttpRequest = async (minIntervalMs: number, signal?: AbortSignal): Promise => { + throwIfAborted(signal); + if (minIntervalMs <= 0) return; + const waitTurn = httpPaceQueue.then(async () => { + throwIfAborted(signal); + const waitMs = + lastHttpRequestStartedAt === undefined + ? 0 + : Math.max(0, lastHttpRequestStartedAt + minIntervalMs - Date.now()); + await abortableSleep(waitMs, signal); + throwIfAborted(signal); + lastHttpRequestStartedAt = Date.now(); + }); + httpPaceQueue = waitTurn.catch(() => undefined); + await waitTurn; +}; + /** * Stable lead of the {@link readConfig} malformed-`GITNEXUS_EMBEDDING_DIMS` * error. `readConfig` throws a plain `Error` (not an {@link HttpEmbeddingError}) @@ -74,6 +152,17 @@ const readConfig = (): HttpConfig | null => { model, apiKey: process.env.GITNEXUS_EMBEDDING_API_KEY ?? 'unused', dimensions, + maxAttempts: parsePositiveIntegerEnv( + 'GITNEXUS_EMBEDDING_MAX_ATTEMPTS', + HTTP_MAX_RETRIES + 1, + 20, + ), + retryCapMs: parsePositiveIntegerEnv( + 'GITNEXUS_EMBEDDING_RETRY_CAP_MS', + HTTP_RETRY_CAP_MS, + 300_000, + ), + minIntervalMs: parseNonNegativeIntegerEnv('GITNEXUS_EMBEDDING_MIN_INTERVAL_MS', 0, 300_000), }; }; @@ -120,11 +209,15 @@ export const safeUrl = (url: string): string => { * its masked form, then strip any residual `scheme://userinfo@` the transport may * have echoed in a normalized (non-exact) form. See #2385. */ -const sanitizeReason = (reason: string, url: string): string => - reason +const sanitizeReason = (reason: string, url: string, apiKey?: string): string => { + const withoutUrlCredentials = reason .split(url) .join(safeUrl(url)) .replace(/([a-z][a-z0-9+.-]*:\/\/)[^/@\s]*@/gi, '$1'); + return apiKey && apiKey !== 'unused' + ? withoutUrlCredentials.split(apiKey).join('[redacted]') + : withoutUrlCredentials; +}; /** * Error thrown by this module's HTTP embedding path (`httpEmbedBatch` / @@ -201,6 +294,10 @@ const httpEmbedBatch = async ( apiKey: string, batchIndex = 0, dimensions?: number, + requestOptions: EmbeddingRequestOptions = {}, + maxAttempts = HTTP_MAX_RETRIES + 1, + retryCapMs = HTTP_RETRY_CAP_MS, + minIntervalMs = 0, ): Promise => { const requestBody: { input: string[]; model: string; dimensions?: number } = { input: batch, @@ -212,11 +309,11 @@ const httpEmbedBatch = async ( let resp: Response; try { + throwIfAborted(requestOptions.signal); resp = await resilientFetch( url, { method: 'POST', - signal: AbortSignal.timeout(HTTP_TIMEOUT_MS), headers: { 'Content-Type': 'application/json', Authorization: `Bearer ${apiKey}`, @@ -224,11 +321,35 @@ const httpEmbedBatch = async ( body: JSON.stringify(requestBody), }, { + fetchImpl: async (input, init) => { + await paceHttpRequest(minIntervalMs, requestOptions.signal); + throwIfAborted(requestOptions.signal); + const timeoutSignal = AbortSignal.timeout(HTTP_TIMEOUT_MS); + const signal = requestOptions.signal + ? AbortSignal.any([requestOptions.signal, timeoutSignal]) + : timeoutSignal; + return globalThis.fetch(input, { ...init, signal }); + }, breakerKey: HTTP_BREAKER_KEY, - retry: { maxAttempts: HTTP_MAX_RETRIES + 1, baseDelayMs: HTTP_RETRY_BACKOFF_MS }, + retry: { + maxAttempts, + baseDelayMs: HTTP_RETRY_BACKOFF_MS, + capDelayMs: retryCapMs, + retryAfterCapMs: retryCapMs, + sleep: (ms) => abortableSleep(ms, requestOptions.signal), + }, }, ); } catch (err) { + if ( + requestOptions.signal?.aborted || + (err instanceof DOMException && err.name === 'AbortError') + ) { + throw new HttpEmbeddingError( + `Embedding request cancelled (${safeUrl(url)}, batch ${batchIndex})`, + { cause: err }, + ); + } if (err instanceof CircuitOpenError) { throw new HttpEmbeddingError( `Embedding endpoint circuit open (${safeUrl(url)}, batch ${batchIndex}): retry in ${Math.ceil(err.retryAfterMs / 1000)}s`, @@ -247,10 +368,12 @@ const httpEmbedBatch = async ( { cause: err }, ); } - const reason = sanitizeReason(err instanceof Error ? err.message : String(err), url); + const reason = sanitizeReason(err instanceof Error ? err.message : String(err), url, apiKey); + const safeCause = new Error(reason); + safeCause.name = err instanceof Error ? err.name : 'EmbeddingTransportError'; throw new HttpEmbeddingError( `Embedding request failed (${safeUrl(url)}, batch ${batchIndex}): ${reason}`, - { cause: err }, + { cause: safeCause }, ); } @@ -290,7 +413,10 @@ const httpEmbedBatch = async ( * @param texts - Array of texts to embed * @returns Array of Float32Array embedding vectors */ -export const httpEmbed = async (texts: string[]): Promise => { +export const httpEmbed = async ( + texts: string[], + requestOptions: EmbeddingRequestOptions = {}, +): Promise => { if (texts.length === 0) return []; const config = readConfig(); @@ -309,6 +435,10 @@ export const httpEmbed = async (texts: string[]): Promise => { config.apiKey, batchIndex, config.dimensions, + requestOptions, + config.maxAttempts, + config.retryCapMs, + config.minIntervalMs, ); if (items.length !== batch.length) { @@ -347,7 +477,10 @@ export const httpEmbed = async (texts: string[]): Promise => { * @param text - Query text to embed * @returns Embedding vector as number array */ -export const httpEmbedQuery = async (text: string): Promise => { +export const httpEmbedQuery = async ( + text: string, + requestOptions: EmbeddingRequestOptions = {}, +): Promise => { const config = readConfig(); if (!config) throw new Error('HTTP embedding not configured'); @@ -359,6 +492,10 @@ export const httpEmbedQuery = async (text: string): Promise => { config.apiKey, 0, config.dimensions, + requestOptions, + config.maxAttempts, + config.retryCapMs, + config.minIntervalMs, ); if (!items.length) { throw new HttpEmbeddingError(`Embedding endpoint returned empty response (${safeUrl(url)})`); diff --git a/gitnexus/src/core/run-analyze.ts b/gitnexus/src/core/run-analyze.ts index c0ed5a06e..9f6143372 100644 --- a/gitnexus/src/core/run-analyze.ts +++ b/gitnexus/src/core/run-analyze.ts @@ -267,6 +267,22 @@ export interface AnalyzeOptions { skipNativeCloseOnExit?: boolean; } +interface EmbeddingIdentity { + model: string; + dimensions: number; +} + +const resolveEmbeddingIdentity = async (): Promise => { + const [{ getEmbeddingDimensions }, { resolveEmbeddingConfig }] = await Promise.all([ + import('./embeddings/embedder.js'), + import('./embeddings/config.js'), + ]); + return { + model: process.env.GITNEXUS_EMBEDDING_MODEL ?? resolveEmbeddingConfig().modelId, + dimensions: getEmbeddingDimensions(), + }; +}; + export interface AnalyzeResult { repoName: string; repoPath: string; @@ -760,6 +776,37 @@ export async function runFullAnalysis( } } + let resumeEmbeddingCheckpoint = false; + let pendingEmbeddingNodeIds = new Set(); + let embeddingIdentityForRun: EmbeddingIdentity | undefined; + if (existingMeta?.embeddingCheckpoint) { + if (options.dropEmbeddings) { + log('Discarding the interrupted embedding checkpoint (--drop-embeddings).'); + options = { ...options, force: true }; + } else { + embeddingIdentityForRun = await resolveEmbeddingIdentity(); + const checkpoint = existingMeta.embeddingCheckpoint; + if ( + checkpoint.model !== embeddingIdentityForRun.model || + checkpoint.dimensions !== embeddingIdentityForRun.dimensions + ) { + throw new Error( + `Cannot resume embedding checkpoint: it uses ${checkpoint.model} at ` + + `${checkpoint.dimensions} dimensions, but this run resolves ` + + `${embeddingIdentityForRun.model} at ${embeddingIdentityForRun.dimensions}. ` + + 'Restore the matching embedding configuration or pass --drop-embeddings to rebuild without it.', + ); + } + resumeEmbeddingCheckpoint = true; + pendingEmbeddingNodeIds = new Set(checkpoint.pendingNodeIds ?? []); + log( + `Previous analyze ended at an embedding checkpoint ` + + `(${checkpoint.nodesProcessed}/${checkpoint.totalNodes} nodes); resuming from persisted hashes` + + `${pendingEmbeddingNodeIds.size > 0 ? ` and regenerating ${pendingEmbeddingNodeIds.size} pending node(s)` : ''}.`, + ); + } + } + // ── Crash recovery: dirty flag forces full rebuild ──────────────── // If the previous incremental run set incrementalInProgress and didn't // clear it, the on-disk index may be in a half-state. Cheapest path @@ -898,7 +945,12 @@ export async function runFullAnalysis( } // ── Early-return: already up to date ────────────────────────────── - if (existingMeta && !options.force && existingMeta.lastCommit === currentCommit) { + if ( + existingMeta && + !existingMeta.embeddingCheckpoint && + !options.force && + existingMeta.lastCommit === currentCommit + ) { // Non-git folders have currentCommit = '' — always rebuild since we can't detect changes if (currentCommit !== '') { // For git repos, even if HEAD matches lastCommit, the working tree @@ -1031,9 +1083,11 @@ export async function runFullAnalysis( const { forceRegenerateEmbeddings, preserveExistingEmbeddings, - shouldGenerateEmbeddings, - shouldLoadCache, + shouldGenerateEmbeddings: derivedShouldGenerateEmbeddings, + shouldLoadCache: derivedShouldLoadCache, } = _deriveEmbeddingMode(options, existingEmbeddingCount); + const shouldGenerateEmbeddings = derivedShouldGenerateEmbeddings || resumeEmbeddingCheckpoint; + const shouldLoadCache = derivedShouldLoadCache || resumeEmbeddingCheckpoint; if (options.dropEmbeddings && existingEmbeddingCount > 0) { log( @@ -1735,7 +1789,7 @@ export async function runFullAnalysis( if (shouldGenerateEmbeddings) { const { skipForCap, capDisabled, nodeLimit } = deriveEmbeddingCap( stats.nodes, - options.embeddingsNodeLimit, + resumeEmbeddingCheckpoint ? 0 : options.embeddingsNodeLimit, ); if (!skipForCap) { embeddingSkipped = false; @@ -1801,6 +1855,8 @@ export async function runFullAnalysis( httpMode ? 'Connecting to embedding endpoint...' : 'Loading embedding model...', ); const { runEmbeddingPipeline } = await import('./embeddings/embedding-pipeline.js'); + embeddingIdentityForRun ??= await resolveEmbeddingIdentity(); + const embeddingIdentity = embeddingIdentityForRun; // Build a Map from cached embeddings for incremental mode let existingEmbeddings: Map | undefined; if (cachedEmbeddingNodeIds.size > 0) { @@ -1810,6 +1866,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, @@ -1826,6 +1924,21 @@ 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( + `MATCH (e:${EMBEDDING_TABLE_NAME}) RETURN count(e) AS cnt`, + ); + const countRow = countResult?.[0]; + const embeddings = Number(countRow?.cnt ?? countRow?.[0] ?? 0); + await saveEmbeddingCheckpoint(checkpoint, [], embeddings); + }, + }, ); if (embeddingResult.semanticMode === 'exact-scan') { semanticMode = 'exact-scan'; @@ -1952,6 +2065,7 @@ export async function runFullAnalysis( // so a sibling branch's prune can union it and not evict our shards. cacheKeys: [...parseCache.usedKeys], incrementalInProgress: undefined as RepoMeta['incrementalInProgress'], + embeddingCheckpoint: undefined, // The effective pdg config this run's DB rows were built under // (#2099 F1). `undefined` on pdg-off runs — this meta is a fresh // literal (no spread of existingMeta), so omission is what CLEARS the diff --git a/gitnexus/src/server/analyze-job.ts b/gitnexus/src/server/analyze-job.ts index d62912abd..5f67fcbfe 100644 --- a/gitnexus/src/server/analyze-job.ts +++ b/gitnexus/src/server/analyze-job.ts @@ -40,6 +40,7 @@ const JOB_TIMEOUT_MS = 30 * 60 * 1000; // 30 minutes export class JobManager { private jobs = new Map(); private children = new Map(); + private abortControllers = new Map(); private timeouts = new Map>(); private emitter = new EventEmitter(); private cleanupTimer: ReturnType; @@ -111,6 +112,7 @@ export class JobManager { if (this.isTerminal(job.status)) { job.completedAt = job.completedAt ?? Date.now(); + this.abortControllers.delete(id); } // Emit exactly one event per updateJob call to prevent SSE double-write @@ -150,6 +152,16 @@ export class JobManager { }); } + /** Register cancellable in-process work for a job. */ + registerAbortController(jobId: string, controller: AbortController): void { + const job = this.jobs.get(jobId); + if (!job || this.isTerminal(job.status)) { + controller.abort(); + return; + } + this.abortControllers.set(jobId, controller); + } + /** Cancel a running job — sends SIGTERM to child process. */ cancelJob(jobId: string, reason?: string): boolean { const job = this.jobs.get(jobId); @@ -159,6 +171,8 @@ export class JobManager { if (child) { child.kill('SIGTERM'); } + this.abortControllers.get(jobId)?.abort(); + this.abortControllers.delete(jobId); this.updateJob(jobId, { status: 'failed', @@ -181,6 +195,8 @@ export class JobManager { child.kill('SIGTERM'); } this.children.clear(); + for (const controller of this.abortControllers.values()) controller.abort(); + this.abortControllers.clear(); // Clear all timeouts for (const timer of this.timeouts.values()) { @@ -201,6 +217,7 @@ export class JobManager { for (const [id, job] of this.jobs) { if (this.isTerminal(job.status) && job.completedAt && now - job.completedAt > JOB_TTL_MS) { this.jobs.delete(id); + this.abortControllers.delete(id); } } } diff --git a/gitnexus/src/server/api.ts b/gitnexus/src/server/api.ts index 218c6e211..50b461f48 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, @@ -1776,17 +1777,15 @@ export const createServer = async (port: number, host: string = '127.0.0.1') => status: 'analyzing' as any, progress: { phase: 'analyzing', percent: 0, message: 'Starting embedding generation...' }, }); + const embedController = new AbortController(); + embedJobManager.registerAbortController(job.id, embedController); // 30-minute timeout for embedding jobs (same as analyze jobs) const EMBED_TIMEOUT_MS = 30 * 60 * 1000; const embedTimeout = setTimeout(() => { const current = embedJobManager.getJob(job.id); if (current && current.status !== 'complete' && current.status !== 'failed') { - releaseRepoLock(repoLockPath); - embedJobManager.updateJob(job.id, { - status: 'failed', - error: 'Embedding timed out (30 minute limit)', - }); + embedJobManager.cancelJob(job.id, 'Embedding timed out (30 minute limit)'); } }, EMBED_TIMEOUT_MS); @@ -1797,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'); @@ -1831,6 +1874,17 @@ export const createServer = async (port: number, host: string = '127.0.0.1') => {}, // config: use defaults undefined, // skipNodeIds existingEmbeddings, + { + signal: embedController.signal, + forceReembedNodeIds, + onCheckpointWindowStart: async ({ nodeIds, ...checkpoint }) => { + await saveEmbeddingCheckpoint(checkpoint, nodeIds); + }, + onCheckpoint: async (checkpoint) => { + await flushWAL(); + await saveEmbeddingCheckpoint(checkpoint, []); + }, + }, ); // Flush WAL so subsequent /api/search requests see the new @@ -1838,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, { @@ -1857,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 4c4e59f18..7755c8c2e 100644 --- a/gitnexus/src/storage/repo-manager.ts +++ b/gitnexus/src/storage/repo-manager.ts @@ -197,6 +197,28 @@ export interface RepoMeta { * under-expanded when the run died. */ droppedImporterChunks?: number; }; + /** + * 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; + nodesProcessed: number; + totalNodes: number; + 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 * default/legacy single-branch case so the flat metadata file stays diff --git a/gitnexus/test/unit/analyze-job.test.ts b/gitnexus/test/unit/analyze-job.test.ts index 4f40d93b9..50b623364 100644 --- a/gitnexus/test/unit/analyze-job.test.ts +++ b/gitnexus/test/unit/analyze-job.test.ts @@ -119,6 +119,17 @@ describe('JobManager', () => { expect(manager.getJob(job.id)!.error).toBe('Cancelled by user'); }); + it('cancelJob aborts registered in-process work', () => { + const job = manager.createJob({ repoPath: '/tmp/repo' }); + manager.updateJob(job.id, { status: 'analyzing' }); + const controller = new AbortController(); + manager.registerAbortController(job.id, controller); + + manager.cancelJob(job.id, 'Cancelled by user'); + + expect(controller.signal.aborted).toBe(true); + }); + it('cancelJob returns false for terminal jobs', () => { const job = manager.createJob({ repoUrl: 'https://github.com/user/repo' }); manager.updateJob(job.id, { status: 'complete' }); diff --git a/gitnexus/test/unit/api-readonly-wiring.test.ts b/gitnexus/test/unit/api-readonly-wiring.test.ts index 7bb8681dc..8e86e7f31 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 570593a2d..18afe0a33 100644 --- a/gitnexus/test/unit/embedding-pipeline.test.ts +++ b/gitnexus/test/unit/embedding-pipeline.test.ts @@ -549,6 +549,189 @@ describe('runEmbeddingPipeline incremental filter', () => { expect(insertN1).toBeLessThan(deleteN2); }); + it('stops at a batch boundary when cancellation is requested', 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 executeQuery = mockExecuteQuery([first, second]); + const executeWithReusedStatement = mockExecuteWithReusedStatement(); + const controller = new AbortController(); + const checkpoints: number[] = []; + + const { runEmbeddingPipeline } = + await import('../../src/core/embeddings/embedding-pipeline.js'); + const promise = runEmbeddingPipeline( + executeQuery, + executeWithReusedStatement, + onProgress, + { batchSize: 1 }, + undefined, + new Map(), + { + signal: controller.signal, + checkpointEveryNodes: 1, + onCheckpoint: async ({ nodesProcessed }) => { + checkpoints.push(nodesProcessed); + controller.abort(); + }, + }, + ); + + await expect(promise).rejects.toThrow(/abort/i); + const insertedIds = stmtCalls + .filter((call) => call.cypher.includes('CREATE')) + .flatMap((call) => call.params.map((param) => param.nodeId)); + expect(insertedIds).toEqual([first.id]); + expect(checkpoints).toEqual([1]); + }); + + it('resumes idempotently from the hashes persisted before an interrupted checkpoint', 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 executeQuery = mockExecuteQuery([first, second]); + const executeWithReusedStatement = mockExecuteWithReusedStatement(); + const { runEmbeddingPipeline } = + await import('../../src/core/embeddings/embedding-pipeline.js'); + + await expect( + runEmbeddingPipeline( + executeQuery, + executeWithReusedStatement, + onProgress, + { batchSize: 1 }, + undefined, + new Map(), + { + checkpointEveryNodes: 1, + onCheckpoint: async ({ nodesProcessed }) => { + if (nodesProcessed === 1) throw new Error('simulated interruption after checkpoint'); + }, + }, + ), + ).rejects.toThrow('simulated interruption'); + + const firstInsert = stmtCalls.find( + (call) => call.cypher.includes('CREATE') && call.params.some((p) => p.nodeId === first.id), + ); + expect(firstInsert).toBeDefined(); + const firstParam = firstInsert?.params.find((param) => param.nodeId === first.id); + if (!firstParam) throw new Error('expected first checkpoint insert'); + const firstHash = firstParam.contentHash; + + stmtCalls = []; + progressUpdates = []; + await runEmbeddingPipeline( + executeQuery, + executeWithReusedStatement, + onProgress, + { batchSize: 1 }, + undefined, + new Map([[first.id, firstHash]]), + { checkpointEveryNodes: 1, onCheckpoint: async () => {} }, + ); + + const resumedIds = stmtCalls + .filter((call) => call.cypher.includes('CREATE')) + .flatMap((call) => call.params.map((param) => param.nodeId)); + 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/http-embedder.test.ts b/gitnexus/test/unit/http-embedder.test.ts index b99cef182..9def24e3b 100644 --- a/gitnexus/test/unit/http-embedder.test.ts +++ b/gitnexus/test/unit/http-embedder.test.ts @@ -6,6 +6,9 @@ const ENV_KEYS = [ 'GITNEXUS_EMBEDDING_MODEL', 'GITNEXUS_EMBEDDING_API_KEY', 'GITNEXUS_EMBEDDING_DIMS', + 'GITNEXUS_EMBEDDING_MAX_ATTEMPTS', + 'GITNEXUS_EMBEDDING_RETRY_CAP_MS', + 'GITNEXUS_EMBEDDING_MIN_INTERVAL_MS', ] as const; /** 384d mock vector matching the default schema dimensions. */ @@ -16,6 +19,7 @@ describe('HTTP embedding backend', () => { const savedEnv = Object.fromEntries(ENV_KEYS.map((k) => [k, process.env[k]])); afterEach(() => { + vi.useRealTimers(); vi.unstubAllGlobals(); vi.resetModules(); // Restore env vars to pre-test state so a mid-test throw can't leak @@ -339,9 +343,25 @@ describe('HTTP embedding backend', () => { expect(isHttpEmbeddingError(err)).toBe(true); // The secret is gone; the masked host is retained so the message stays useful. expect(String(err)).not.toContain('secret'); + expect(String((err as Error & { cause?: unknown }).cause)).not.toContain('secret'); expect(String(err)).toContain('host.example'); }); + it('redacts the API key from both the message and diagnostic cause', async () => { + process.env.GITNEXUS_EMBEDDING_URL = 'https://host.example/v1'; + process.env.GITNEXUS_EMBEDDING_MODEL = 'test-model'; + process.env.GITNEXUS_EMBEDDING_API_KEY = 'super-secret-key'; + vi.stubGlobal( + 'fetch', + vi.fn().mockRejectedValue(new TypeError('transport rejected super-secret-key')), + ); + + const { embedText } = await import('../../src/core/embeddings/embedder.js'); + const err = await embedText('test').catch((error: unknown) => error); + expect(String(err)).not.toContain('super-secret-key'); + expect(String((err as Error & { cause?: unknown }).cause)).not.toContain('super-secret-key'); + }); + it('leaves a non-credential reason unchanged (no over-scrubbing)', async () => { process.env.GITNEXUS_EMBEDDING_URL = 'http://test:8080/v1'; process.env.GITNEXUS_EMBEDDING_MODEL = 'test-model'; @@ -563,6 +583,114 @@ describe('HTTP embedding backend', () => { expect(fetch).toHaveBeenCalledTimes(2); expect(result).toBeInstanceOf(Float32Array); }); + + it('honors the configured total attempt bound', async () => { + process.env.GITNEXUS_EMBEDDING_URL = 'http://test:8080/v1'; + process.env.GITNEXUS_EMBEDDING_MODEL = 'test-model'; + process.env.GITNEXUS_EMBEDDING_MAX_ATTEMPTS = '1'; + vi.stubGlobal('fetch', vi.fn().mockResolvedValue({ ok: false, status: 503 })); + + const { embedText } = await import('../../src/core/embeddings/embedder.js'); + await expect(embedText('test')).rejects.toThrow('503'); + expect(fetch).toHaveBeenCalledTimes(1); + }); + + it('caps Retry-After with the configured retry cap', async () => { + vi.useFakeTimers(); + vi.setSystemTime(0); + process.env.GITNEXUS_EMBEDDING_URL = 'http://test:8080/v1'; + process.env.GITNEXUS_EMBEDDING_MODEL = 'test-model'; + process.env.GITNEXUS_EMBEDDING_MAX_ATTEMPTS = '2'; + process.env.GITNEXUS_EMBEDDING_RETRY_CAP_MS = '2500'; + const ok = { ok: true, json: async () => ({ data: [{ embedding: mockVec }] }) }; + vi.stubGlobal( + 'fetch', + vi + .fn() + .mockResolvedValueOnce( + new Response('{}', { status: 429, headers: { 'Retry-After': '60' } }), + ) + .mockResolvedValueOnce(ok), + ); + + const { embedText } = await import('../../src/core/embeddings/embedder.js'); + const promise = embedText('test'); + await vi.advanceTimersByTimeAsync(2499); + expect(fetch).toHaveBeenCalledTimes(1); + await vi.advanceTimersByTimeAsync(1); + await expect(promise).resolves.toBeInstanceOf(Float32Array); + expect(fetch).toHaveBeenCalledTimes(2); + }); + + it('paces retries and successful batches through one minimum-interval queue', async () => { + vi.useFakeTimers(); + vi.setSystemTime(0); + process.env.GITNEXUS_EMBEDDING_URL = 'http://test:8080/v1'; + process.env.GITNEXUS_EMBEDDING_MODEL = 'test-model'; + process.env.GITNEXUS_EMBEDDING_MAX_ATTEMPTS = '2'; + process.env.GITNEXUS_EMBEDDING_RETRY_CAP_MS = '1'; + process.env.GITNEXUS_EMBEDDING_MIN_INTERVAL_MS = '1000'; + const makeResp = (count: number) => ({ + ok: true, + json: async () => ({ data: Array.from({ length: count }, () => ({ embedding: mockVec })) }), + }); + vi.stubGlobal( + 'fetch', + vi + .fn() + .mockResolvedValueOnce({ ok: false, status: 503 }) + .mockResolvedValueOnce(makeResp(64)) + .mockResolvedValueOnce(makeResp(6)), + ); + + const { embedBatch } = await import('../../src/core/embeddings/embedder.js'); + const promise = embedBatch(Array.from({ length: 70 }, (_, i) => `text ${i}`)); + await vi.advanceTimersByTimeAsync(0); + expect(fetch).toHaveBeenCalledTimes(1); + await vi.advanceTimersByTimeAsync(999); + expect(fetch).toHaveBeenCalledTimes(1); + await vi.advanceTimersByTimeAsync(1); + expect(fetch).toHaveBeenCalledTimes(2); + await vi.advanceTimersByTimeAsync(999); + expect(fetch).toHaveBeenCalledTimes(2); + await vi.advanceTimersByTimeAsync(1); + await expect(promise).resolves.toHaveLength(70); + expect(fetch).toHaveBeenCalledTimes(3); + }); + + it('cancels promptly while waiting for retry backoff', async () => { + vi.useFakeTimers(); + process.env.GITNEXUS_EMBEDDING_URL = 'http://test:8080/v1'; + process.env.GITNEXUS_EMBEDDING_MODEL = 'test-model'; + process.env.GITNEXUS_EMBEDDING_MAX_ATTEMPTS = '3'; + process.env.GITNEXUS_EMBEDDING_RETRY_CAP_MS = '60000'; + vi.stubGlobal( + 'fetch', + vi + .fn() + .mockResolvedValue(new Response('{}', { status: 429, headers: { 'Retry-After': '60' } })), + ); + const controller = new AbortController(); + + const { embedBatch } = await import('../../src/core/embeddings/embedder.js'); + const promise = embedBatch(['test'], { signal: controller.signal }); + await vi.advanceTimersByTimeAsync(1); + controller.abort(); + await expect(promise).rejects.toThrow(/cancelled/i); + expect(fetch).toHaveBeenCalledTimes(1); + }); + + it.each([ + ['GITNEXUS_EMBEDDING_MAX_ATTEMPTS', '0'], + ['GITNEXUS_EMBEDDING_RETRY_CAP_MS', '-1'], + ['GITNEXUS_EMBEDDING_MIN_INTERVAL_MS', 'nope'], + ])('rejects malformed resilience config %s=%s', async (key, value) => { + process.env.GITNEXUS_EMBEDDING_URL = 'http://test:8080/v1'; + process.env.GITNEXUS_EMBEDDING_MODEL = 'test-model'; + process.env[key] = value; + const { embedText } = await import('../../src/core/embeddings/embedder.js'); + await expect(embedText('test')).rejects.toThrow(key); + }); }); describe('dimension mismatch on query path', () => { diff --git a/gitnexus/test/unit/integrations/resilient-fetch.test.ts b/gitnexus/test/unit/integrations/resilient-fetch.test.ts index 05299c0ba..fa7114e04 100644 --- a/gitnexus/test/unit/integrations/resilient-fetch.test.ts +++ b/gitnexus/test/unit/integrations/resilient-fetch.test.ts @@ -185,6 +185,22 @@ describe('resilientFetch', () => { expect(sleep).toHaveBeenCalledWith(50); }); + it('lets a caller set a stricter Retry-After cap', async () => { + let n = 0; + const fetchImpl = vi.fn(async () => { + n += 1; + return n === 1 ? jsonResp(429, { 'Retry-After': '60' }) : jsonResp(204); + }); + const sleep = vi.fn(async () => {}); + const { breaker } = makeBreaker(); + await resilientFetch(URL_STR, undefined, { + fetchImpl: fetchImpl as unknown as typeof fetch, + breaker, + retry: { sleep, capDelayMs: 2500, retryAfterCapMs: 2500 }, + }); + expect(sleep).toHaveBeenCalledWith(2500); + }); + it('401 returned as Response, no retry, breaker not incremented', async () => { const fetchImpl = vi.fn(async () => jsonResp(401)); const sleep = vi.fn(async () => {}); diff --git a/gitnexus/test/unit/run-analyze.test.ts b/gitnexus/test/unit/run-analyze.test.ts index 93a2f8915..af2c64ad8 100644 --- a/gitnexus/test/unit/run-analyze.test.ts +++ b/gitnexus/test/unit/run-analyze.test.ts @@ -1,7 +1,7 @@ import { execSync } from 'child_process'; import fs from 'fs/promises'; import path from 'path'; -import { describe, it, expect } from 'vitest'; +import { describe, it, expect, vi } from 'vitest'; import { deriveEmbeddingMode, deriveEmbeddingCap, @@ -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 () => { @@ -74,6 +75,145 @@ describe('run-analyze module', () => { } }); + it('resumes a matching embedding checkpoint instead of taking the clean fast path', async () => { + const tmpRepo = await createTempDir('gitnexus-run-analyze-embedding-checkpoint-'); + const tmpHome = await createTempDir('gitnexus-run-analyze-embedding-checkpoint-home-'); + const saved = { + home: process.env.GITNEXUS_HOME, + url: process.env.GITNEXUS_EMBEDDING_URL, + model: process.env.GITNEXUS_EMBEDDING_MODEL, + dims: process.env.GITNEXUS_EMBEDDING_DIMS, + extension: process.env.GITNEXUS_LBUG_EXTENSION_INSTALL, + }; + try { + process.env.GITNEXUS_HOME = tmpHome.dbPath; + process.env.GITNEXUS_EMBEDDING_URL = 'http://test:8080/v1'; + process.env.GITNEXUS_EMBEDDING_MODEL = 'test-model'; + process.env.GITNEXUS_EMBEDDING_DIMS = '384'; + process.env.GITNEXUS_LBUG_EXTENSION_INSTALL = 'never'; + const vector = Array.from({ length: 384 }, (_, i) => i / 384); + const fetchMock = vi.fn().mockImplementation(async (_input, init?: RequestInit) => { + const body = JSON.parse(String(init?.body ?? '{}')) as { input?: unknown[] }; + const count = Array.isArray(body.input) ? body.input.length : 1; + return { + ok: true, + json: async () => ({ + data: Array.from({ length: count }, () => ({ embedding: vector })), + }), + }; + }); + vi.stubGlobal('fetch', fetchMock); + await fs.writeFile( + path.join(tmpRepo.dbPath, 'index.ts'), + 'export function checkpointResume() { return "ready"; }\n', + ); + execSync('git init', { cwd: tmpRepo.dbPath, stdio: 'pipe' }); + execSync('git add index.ts', { cwd: tmpRepo.dbPath, stdio: 'pipe' }); + execSync('git -c user.name=test -c user.email=test@test commit -m init', { + cwd: tmpRepo.dbPath, + stdio: 'pipe', + }); + + const { runFullAnalysis } = await import('../../src/core/run-analyze.js'); + await runFullAnalysis( + tmpRepo.dbPath, + { embeddings: true, skipAgentsMd: true, skipSkills: true }, + { onProgress: () => {} }, + ); + const { storagePath } = getStoragePaths(tmpRepo.dbPath); + const completed = await loadMeta(storagePath); + expect(completed).not.toBeNull(); + if (!completed) throw new Error('expected completed metadata'); + await saveMeta(storagePath, { + ...completed, + embeddingCheckpoint: { + at: new Date().toISOString(), + nodesProcessed: 1, + totalNodes: 1, + chunksProcessed: 1, + model: 'test-model', + dimensions: 384, + }, + } as RepoMeta); + fetchMock.mockClear(); + const logs: string[] = []; + + const resumed = await runFullAnalysis( + tmpRepo.dbPath, + { skipAgentsMd: true, skipSkills: true }, + { onProgress: () => {}, onLog: (message) => logs.push(message) }, + ); + + expect(resumed.alreadyUpToDate).not.toBe(true); + expect(fetchMock).not.toHaveBeenCalled(); + expect(logs.some((message) => message.includes('embedding checkpoint'))).toBe(true); + expect((await loadMeta(storagePath))?.embeddingCheckpoint).toBeUndefined(); + + 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, + totalNodes: 2, + chunksProcessed: 1, + model: 'different-model', + dimensions: 384, + }, + }); + await expect( + runFullAnalysis( + tmpRepo.dbPath, + { skipAgentsMd: true, skipSkills: true }, + { onProgress: () => {} }, + ), + ).rejects.toThrow('Cannot resume embedding checkpoint'); + expect(fetchMock).not.toHaveBeenCalled(); + } finally { + vi.unstubAllGlobals(); + const restore = (key: string, value: string | undefined) => { + if (value === undefined) delete process.env[key]; + else process.env[key] = value; + }; + restore('GITNEXUS_HOME', saved.home); + restore('GITNEXUS_EMBEDDING_URL', saved.url); + restore('GITNEXUS_EMBEDDING_MODEL', saved.model); + restore('GITNEXUS_EMBEDDING_DIMS', saved.dims); + restore('GITNEXUS_LBUG_EXTENSION_INSTALL', saved.extension); + await tmpRepo.cleanup(); + await tmpHome.cleanup(); + } + }, 120_000); + it('plain analyze on another branch adopts the flat workspace slot (#2354)', async () => { const tmpRepo = await createTempDir('gitnexus-run-analyze-workspace-'); const tmpHome = await createTempDir('gitnexus-run-analyze-workspace-home-'); From c35fc27427ae8cce10ad2b0ac03ec954812eff5f Mon Sep 17 00:00:00 2001 From: Eva Date: Tue, 14 Jul 2026 03:07:33 +0700 Subject: [PATCH 3/9] fix(cli): fail cypher errors loudly --- gitnexus/src/cli/tool.ts | 9 +++++++++ gitnexus/test/unit/tool-direct-cli.test.ts | 23 ++++++++++++++++++++++ 2 files changed, 32 insertions(+) diff --git a/gitnexus/src/cli/tool.ts b/gitnexus/src/cli/tool.ts index 688c66535..46adaf990 100644 --- a/gitnexus/src/cli/tool.ts +++ b/gitnexus/src/cli/tool.ts @@ -301,6 +301,15 @@ export async function cypherCommand( } } output(result); + if ( + result && + typeof result === 'object' && + 'error' in result && + typeof result.error === 'string' && + result.error.trim().length > 0 + ) { + process.exitCode = 1; + } } export async function detectChangesCommand(options?: { diff --git a/gitnexus/test/unit/tool-direct-cli.test.ts b/gitnexus/test/unit/tool-direct-cli.test.ts index 62e5fa1ea..2230a675e 100644 --- a/gitnexus/test/unit/tool-direct-cli.test.ts +++ b/gitnexus/test/unit/tool-direct-cli.test.ts @@ -81,6 +81,29 @@ describe('direct CLI tool commands', () => { expect(process.exitCode).toBe(1); }); + it('fails closed when cypher returns a backend error payload', async () => { + callToolMock.mockResolvedValue({ error: 'Binder exception: missing relationship property' }); + const { cypherCommand } = await import('../../src/cli/tool.js'); + + await cypherCommand('MATCH ()-[r:CodeRelation]->() RETURN r.missing'); + + expect(writeSyncMock).toHaveBeenCalledWith( + 1, + expect.stringContaining('Binder exception: missing relationship property'), + ); + expect(process.exitCode).toBe(1); + }); + + it('keeps a successful cypher result at exit zero', async () => { + callToolMock.mockResolvedValue({ markdown: '| count |\n| --- |\n| 1 |', row_count: 1 }); + const { cypherCommand } = await import('../../src/cli/tool.js'); + + await cypherCommand('MATCH (n) RETURN count(n) AS count'); + + expect(writeSyncMock).toHaveBeenCalledWith(1, expect.stringContaining('"row_count": 1')); + expect(process.exitCode).toBeUndefined(); + }); + it('dispatches detect_changes with CLI-shaped arguments', async () => { callToolMock.mockResolvedValue({ summary: { From 3d6908ba4b2cf15a9c171ab0acf2e0790b920f63 Mon Sep 17 00:00:00 2001 From: Eva Date: Tue, 14 Jul 2026 03:17:08 +0700 Subject: [PATCH 4/9] fix(cache): bound parsedfile generations --- .../ingestion/pipeline-phases/parse-impl.ts | 4 ++ gitnexus/src/storage/parse-cache.ts | 2 +- gitnexus/src/storage/parsedfile-store.ts | 16 +++++++ ...mpl-warm-cache-parsedfile-coverage.test.ts | 45 +++++++++++++++++++ 4 files changed, 66 insertions(+), 1 deletion(-) diff --git a/gitnexus/src/core/ingestion/pipeline-phases/parse-impl.ts b/gitnexus/src/core/ingestion/pipeline-phases/parse-impl.ts index 257a04bc8..b557625eb 100644 --- a/gitnexus/src/core/ingestion/pipeline-phases/parse-impl.ts +++ b/gitnexus/src/core/ingestion/pipeline-phases/parse-impl.ts @@ -33,6 +33,7 @@ import { persistParsedFileChunk, getDurableParsedFileDir, loadDurableParsedFileIndex, + prepareDurableParsedFileChunk, restoreDurableParsedFileShard, } from '../../../storage/parsedfile-store.js'; import type { ParseWorkerResult } from '../workers/parse-worker.js'; @@ -984,6 +985,9 @@ export async function runChunkedParseAndResolve( // Cache miss: dispatch to workers, capture the raw results, store // them under the chunk hash for the next run. chunkCacheMisses++; + if (durableParsedFileDir !== undefined && chunkHash !== null) { + await prepareDurableParsedFileChunk(durableParsedFileDir, chunkHash); + } const progressForChunk = (current: number, _total: number, filePath: string) => { const globalCurrent = filesParsedSoFar + current; // Parse phase covers 20-70 (M2). Deferred extraction handles 70-95. diff --git a/gitnexus/src/storage/parse-cache.ts b/gitnexus/src/storage/parse-cache.ts index 1353f6324..b11b8c27f 100644 --- a/gitnexus/src/storage/parse-cache.ts +++ b/gitnexus/src/storage/parse-cache.ts @@ -55,7 +55,7 @@ import type { ParseWorkerResult } from '../core/ingestion/workers/parse-worker.j // the main thread (the #1983 OOM). Because the two stores share this version, // any future change to the `ParsedFile` serialization shape MUST bump // SCHEMA_BUMP so both invalidate in lockstep. -const SCHEMA_BUMP = 12; // #2391 follow-up: extractPythonModuleConstants changed what it EMITS for the same source (binding mutual-exclusivity clears stale imports; RHS refs are snapshotted; `$imp$N` aliases). `moduleConstants` is cached verbatim, so a warm shard built pre-fix would replay stale/WRONG folds and the correctness fixes would silently no-op on upgrade — bump to force re-extraction. (11 = #2391: ExtractedDecoratorRoute gained `routePathExpr`/`routePathOperands` + ParseWorkerResult gained per-file `moduleConstants`. 10 = PR #2200: Property nodes gained `rawDeclaredType` + `annotations` for Spring DI) +const SCHEMA_BUMP = 13; // Durable ParsedFile chunk directories now replace one complete generation instead of accumulating worker shards across cache-miss analyses. Invalidate once so existing unbounded stores are rebuilt under the bounded contract. (12 = #2391 follow-up: Python module constant extraction semantics changed.) const GITNEXUS_PKG_VERSION = (() => { try { // package.json sits at gitnexus/package.json — two levels up from diff --git a/gitnexus/src/storage/parsedfile-store.ts b/gitnexus/src/storage/parsedfile-store.ts index 842be5c2e..302202c5d 100644 --- a/gitnexus/src/storage/parsedfile-store.ts +++ b/gitnexus/src/storage/parsedfile-store.ts @@ -292,6 +292,22 @@ export const getDurableParsedFileDir = (storagePath: string): string => const durableChunkDir = (durableDir: string, chunkHash: string): string => path.join(durableDir, chunkHash); +/** + * Start a fresh durable generation for one content-addressed parse chunk. + * The main thread calls this once before dispatching a cache miss, before any + * worker can write that chunk. Recreating the directory immediately keeps the + * worker-side mkdir memoization valid while preventing old worker shard names + * from accumulating across analyses. + */ +export const prepareDurableParsedFileChunk = async ( + durableDir: string, + chunkHash: string, +): Promise => { + const dir = durableChunkDir(durableDir, chunkHash); + await fs.rm(dir, { recursive: true, force: true }); + await fs.mkdir(dir, { recursive: true }); +}; + // Per-process set of durable chunk subdirs already `mkdir`ed (mirrors // `createdStoreDirs`) so the worker doesn't `mkdirSync` on every shard. const createdDurableDirs = new Set(); diff --git a/gitnexus/test/unit/parse-impl-warm-cache-parsedfile-coverage.test.ts b/gitnexus/test/unit/parse-impl-warm-cache-parsedfile-coverage.test.ts index e98083737..dabca8001 100644 --- a/gitnexus/test/unit/parse-impl-warm-cache-parsedfile-coverage.test.ts +++ b/gitnexus/test/unit/parse-impl-warm-cache-parsedfile-coverage.test.ts @@ -41,6 +41,7 @@ import { } from '../../src/storage/parse-cache.js'; import { getDurableParsedFileDir, + prepareDurableParsedFileChunk, persistDurableParsedFileShardSync, restoreDurableParsedFileShard, loadParsedFilesForPaths, @@ -100,6 +101,29 @@ describe('durable ParsedFile store — content-addressed warm-cache coverage', ( expect(restored).toBe(0); }); + it('prepares a fresh durable generation without retaining old worker shards', async () => { + const durableDir = getDurableParsedFileDir(tempDir); + const chunkHash = 'f'.repeat(64); + const chunkDir = path.join(durableDir, chunkHash); + + persistDurableParsedFileShardSync(durableDir, chunkHash, 1, 0, [mkParsedFile('old.ts')]); + await prepareDurableParsedFileChunk(durableDir, chunkHash); + persistDurableParsedFileShardSync(durableDir, chunkHash, 1, 0, [mkParsedFile('new-a.ts')]); + persistDurableParsedFileShardSync(durableDir, chunkHash, 2, 0, [mkParsedFile('new-b.ts')]); + + const shards = fs + .readdirSync(chunkDir) + .filter((name) => name.endsWith('.json')) + .sort(); + expect(shards).toEqual([`${chunkHash}-w1-0.json`, `${chunkHash}-w2-0.json`]); + await restoreDurableParsedFileShard(durableDir, tempDir, chunkHash); + const files = await loadParsedFilesForPaths( + tempDir, + new Set(['old.ts', 'new-a.ts', 'new-b.ts']), + ); + expect([...files.keys()].sort()).toEqual(['new-a.ts', 'new-b.ts']); + }); + it('index load is version-gated (PARSE_CACHE_VERSION mismatch ⇒ empty)', async () => { const durableDir = getDurableParsedFileDir(tempDir); const chunkHash = 'c'.repeat(64); @@ -291,6 +315,27 @@ describe('parse-impl warm-cache ParsedFile coverage (#2038)', () => { expect(cache.usedKeys.has(chunkHash)).toBe(true); }); + it('a repeated cache miss replaces the durable chunk generation', async () => { + const f = writeFile('src/repeated.ts', 'export function repeated() { return 1; }\n'); + const chunkHash = computeChunkHash([ + { + filePath: f.path, + contentHash: fileContentHash(fs.readFileSync(path.join(repoDir, f.path), 'utf-8')), + }, + ]); + + await run(newCache(), [f]); + await run(newCache(), [f]); + + const chunkDir = path.join(getDurableParsedFileDir(storageDir), chunkHash); + const shards = fs.readdirSync(chunkDir).filter((name) => name.endsWith('.json')); + expect(shards).toHaveLength(1); + const parsed = JSON.parse(fs.readFileSync(path.join(chunkDir, shards[0]!), 'utf-8')) as Array<{ + filePath: string; + }>; + expect(parsed.map((item) => item.filePath)).toEqual(['src/repeated.ts']); + }); + it('run #2 (all hits) spawns NO worker — the warm path is served from caches', async () => { const f = writeFile('src/cached.ts', 'export function cached() { return 1; }\n'); const cache = newCache(); From 1821b01dbec71f8eeede6060316464d4ea31a093 Mon Sep 17 00:00:00 2001 From: Eva Date: Thu, 16 Jul 2026 09:48:03 +0700 Subject: [PATCH 5/9] fix(embeddings): bind resume checkpoints to provider --- .../src/core/embeddings/embedding-identity.ts | 32 +++++++++++++++++++ gitnexus/src/core/run-analyze.ts | 32 ++++++++----------- gitnexus/src/server/api.ts | 16 +++++----- gitnexus/src/storage/repo-manager.ts | 2 ++ gitnexus/test/unit/http-embedder.test.ts | 21 ++++++++++++ gitnexus/test/unit/run-analyze.test.ts | 27 ++++++++++++++++ 6 files changed, 104 insertions(+), 26 deletions(-) create mode 100644 gitnexus/src/core/embeddings/embedding-identity.ts diff --git a/gitnexus/src/core/embeddings/embedding-identity.ts b/gitnexus/src/core/embeddings/embedding-identity.ts new file mode 100644 index 000000000..fa185c192 --- /dev/null +++ b/gitnexus/src/core/embeddings/embedding-identity.ts @@ -0,0 +1,32 @@ +import { createHash } from 'node:crypto'; +import { getEmbeddingDimensions } from './embedder.js'; +import { resolveEmbeddingConfig } from './config.js'; +import { isHttpMode, safeUrl } from './http-client.js'; + +export interface EmbeddingIdentity { + model: string; + dimensions: number; + provider: string; +} + +/** + * Identify the vector space strongly enough to resume without mixing providers. + * The HTTP fingerprint excludes URL credentials and query parameters before + * hashing, so metadata contains neither an endpoint nor a secret-derived hash. + */ +export function resolveEmbeddingIdentity(): EmbeddingIdentity { + const httpMode = isHttpMode(); + const provider = httpMode + ? `http:${createHash('sha256') + .update(safeUrl(process.env.GITNEXUS_EMBEDDING_URL ?? '')) + .digest('hex')}` + : 'local'; + + return { + model: httpMode + ? (process.env.GITNEXUS_EMBEDDING_MODEL as string) + : resolveEmbeddingConfig().modelId, + dimensions: getEmbeddingDimensions(), + provider, + }; +} diff --git a/gitnexus/src/core/run-analyze.ts b/gitnexus/src/core/run-analyze.ts index d5b1bc7c4..18f9f37aa 100644 --- a/gitnexus/src/core/run-analyze.ts +++ b/gitnexus/src/core/run-analyze.ts @@ -53,6 +53,7 @@ import { type WalCheckpointDriver, } from './lbug/wal-checkpoint-driver.js'; import { quarantineSidecarsForDirtyRecovery } from './lbug/sidecar-recovery.js'; +import type { EmbeddingIdentity } from './embeddings/embedding-identity.js'; import { getStoragePaths, resolveBranchPlacement, @@ -267,22 +268,6 @@ export interface AnalyzeOptions { skipNativeCloseOnExit?: boolean; } -interface EmbeddingIdentity { - model: string; - dimensions: number; -} - -const resolveEmbeddingIdentity = async (): Promise => { - const [{ getEmbeddingDimensions }, { resolveEmbeddingConfig }] = await Promise.all([ - import('./embeddings/embedder.js'), - import('./embeddings/config.js'), - ]); - return { - model: process.env.GITNEXUS_EMBEDDING_MODEL ?? resolveEmbeddingConfig().modelId, - dimensions: getEmbeddingDimensions(), - }; -}; - export interface AnalyzeResult { repoName: string; repoPath: string; @@ -784,8 +769,15 @@ export async function runFullAnalysis( log('Discarding the interrupted embedding checkpoint (--drop-embeddings).'); options = { ...options, force: true }; } else { - embeddingIdentityForRun = await resolveEmbeddingIdentity(); + const { resolveEmbeddingIdentity } = await import('./embeddings/embedding-identity.js'); + embeddingIdentityForRun = resolveEmbeddingIdentity(); const checkpoint = existingMeta.embeddingCheckpoint; + if (checkpoint.provider !== embeddingIdentityForRun.provider) { + throw new Error( + 'Cannot resume embedding checkpoint: the embedding provider configuration differs. ' + + 'Restore the matching endpoint configuration or pass --drop-embeddings to rebuild without it.', + ); + } if ( checkpoint.model !== embeddingIdentityForRun.model || checkpoint.dimensions !== embeddingIdentityForRun.dimensions @@ -1855,7 +1847,10 @@ export async function runFullAnalysis( httpMode ? 'Connecting to embedding endpoint...' : 'Loading embedding model...', ); const { runEmbeddingPipeline } = await import('./embeddings/embedding-pipeline.js'); - embeddingIdentityForRun ??= await resolveEmbeddingIdentity(); + if (!embeddingIdentityForRun) { + const { resolveEmbeddingIdentity } = await import('./embeddings/embedding-identity.js'); + embeddingIdentityForRun = resolveEmbeddingIdentity(); + } const embeddingIdentity = embeddingIdentityForRun; // Build a Map from cached embeddings for incremental mode let existingEmbeddings: Map | undefined; @@ -1902,6 +1897,7 @@ export async function runFullAnalysis( ...checkpoint, model: embeddingIdentity.model, dimensions: embeddingIdentity.dimensions, + provider: embeddingIdentity.provider, pendingNodeIds, }, pdg: resolvePdgConfig(options), diff --git a/gitnexus/src/server/api.ts b/gitnexus/src/server/api.ts index 50b461f48..aefc15246 100644 --- a/gitnexus/src/server/api.ts +++ b/gitnexus/src/server/api.ts @@ -1796,19 +1796,19 @@ 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(), - }; + const { resolveEmbeddingIdentity } = + await import('../core/embeddings/embedding-identity.js'); + const embeddingIdentity = resolveEmbeddingIdentity(); 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.provider !== embeddingIdentity.provider) { + throw new Error( + 'Cannot resume embedding checkpoint: the embedding provider configuration differs.', + ); + } if ( priorCheckpoint && (priorCheckpoint.model !== embeddingIdentity.model || diff --git a/gitnexus/src/storage/repo-manager.ts b/gitnexus/src/storage/repo-manager.ts index 7755c8c2e..9b3751898 100644 --- a/gitnexus/src/storage/repo-manager.ts +++ b/gitnexus/src/storage/repo-manager.ts @@ -211,6 +211,8 @@ export interface RepoMeta { chunksProcessed: number; model: string; dimensions: number; + /** `local` or a secret-free SHA-256 fingerprint of the HTTP endpoint identity. */ + provider: string; /** * Nodes in the current checkpoint window. Any of these may have only a * subset of their chunks persisted after an abrupt process termination, diff --git a/gitnexus/test/unit/http-embedder.test.ts b/gitnexus/test/unit/http-embedder.test.ts index 9def24e3b..63cd715f9 100644 --- a/gitnexus/test/unit/http-embedder.test.ts +++ b/gitnexus/test/unit/http-embedder.test.ts @@ -32,6 +32,27 @@ describe('HTTP embedding backend', () => { } }); + it('fingerprints HTTP provider identity without confusing a model-only env with HTTP mode', async () => { + process.env.GITNEXUS_EMBEDDING_URL = 'https://user:secret@first.example/v1?token=hidden'; + process.env.GITNEXUS_EMBEDDING_MODEL = 'shared-model-name'; + process.env.GITNEXUS_EMBEDDING_DIMS = '384'; + const { resolveEmbeddingIdentity } = + await import('../../src/core/embeddings/embedding-identity.js'); + + const first = resolveEmbeddingIdentity(); + process.env.GITNEXUS_EMBEDDING_URL = 'https://second.example/v1'; + const second = resolveEmbeddingIdentity(); + delete process.env.GITNEXUS_EMBEDDING_URL; + const local = resolveEmbeddingIdentity(); + + expect(first.provider).toMatch(/^http:[0-9a-f]{64}$/u); + expect(first.provider).not.toContain('secret'); + expect(first.provider).not.toContain('hidden'); + expect(second.provider).not.toBe(first.provider); + expect(local.provider).toBe('local'); + expect(local.model).not.toBe('shared-model-name'); + }); + describe('MCP embedder', () => { it('returns 384 dimensions by default', () => { expect(getEmbeddingDims()).toBe(384); diff --git a/gitnexus/test/unit/run-analyze.test.ts b/gitnexus/test/unit/run-analyze.test.ts index af2c64ad8..7beea1243 100644 --- a/gitnexus/test/unit/run-analyze.test.ts +++ b/gitnexus/test/unit/run-analyze.test.ts @@ -124,6 +124,9 @@ describe('run-analyze module', () => { const completed = await loadMeta(storagePath); expect(completed).not.toBeNull(); if (!completed) throw new Error('expected completed metadata'); + const { resolveEmbeddingIdentity } = + await import('../../src/core/embeddings/embedding-identity.js'); + const embeddingIdentity = resolveEmbeddingIdentity(); await saveMeta(storagePath, { ...completed, embeddingCheckpoint: { @@ -133,6 +136,7 @@ describe('run-analyze module', () => { chunksProcessed: 1, model: 'test-model', dimensions: 384, + provider: embeddingIdentity.provider, }, } as RepoMeta); fetchMock.mockClear(); @@ -162,6 +166,7 @@ describe('run-analyze module', () => { chunksProcessed: 0, model: 'test-model', dimensions: 384, + provider: embeddingIdentity.provider, pendingNodeIds: [pendingNodeId], }, }); @@ -179,6 +184,27 @@ describe('run-analyze module', () => { 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, + totalNodes: 2, + chunksProcessed: 1, + model: 'test-model', + dimensions: 384, + provider: 'http:different-provider-fingerprint', + }, + }); + await expect( + runFullAnalysis( + tmpRepo.dbPath, + { skipAgentsMd: true, skipSkills: true }, + { onProgress: () => {} }, + ), + ).rejects.toThrow(/provider configuration differs/i); + expect(fetchMock).not.toHaveBeenCalled(); + await saveMeta(storagePath, { ...resumedPending, embeddingCheckpoint: { @@ -188,6 +214,7 @@ describe('run-analyze module', () => { chunksProcessed: 1, model: 'different-model', dimensions: 384, + provider: embeddingIdentity.provider, }, }); await expect( From ee7161ef0d9bd186b8e3b44b034ab185664fb9b7 Mon Sep 17 00:00:00 2001 From: Gergo Magyar Date: Thu, 16 Jul 2026 09:55:44 +0000 Subject: [PATCH 6/9] fix(cache): degrade when the durable generation reset fails An fs failure while resetting a chunk generation now warns and continues like the neighboring durable-store paths instead of failing the analyze. Workers recreate the directory on write. Co-Authored-By: Claude Fable 5 --- .../ingestion/pipeline-phases/parse-impl.ts | 12 +++++++- ...mpl-warm-cache-parsedfile-coverage.test.ts | 28 ++++++++++++++++++- 2 files changed, 38 insertions(+), 2 deletions(-) diff --git a/gitnexus/src/core/ingestion/pipeline-phases/parse-impl.ts b/gitnexus/src/core/ingestion/pipeline-phases/parse-impl.ts index b557625eb..ad72f5506 100644 --- a/gitnexus/src/core/ingestion/pipeline-phases/parse-impl.ts +++ b/gitnexus/src/core/ingestion/pipeline-phases/parse-impl.ts @@ -986,7 +986,17 @@ export async function runChunkedParseAndResolve( // them under the chunk hash for the next run. chunkCacheMisses++; if (durableParsedFileDir !== undefined && chunkHash !== null) { - await prepareDurableParsedFileChunk(durableParsedFileDir, chunkHash); + try { + await prepareDurableParsedFileChunk(durableParsedFileDir, chunkHash); + } catch (err) { + // The durable store is an optimization — degrade like the restore + // path does instead of failing the analyze. Workers recreate the + // directory on write, so at worst the old generation lingers. + logger.warn( + { err, chunkHash: chunkHash.slice(0, 8) }, + 'parsedfile-cache: could not reset durable chunk generation; continuing', + ); + } } const progressForChunk = (current: number, _total: number, filePath: string) => { const globalCurrent = filesParsedSoFar + current; diff --git a/gitnexus/test/unit/parse-impl-warm-cache-parsedfile-coverage.test.ts b/gitnexus/test/unit/parse-impl-warm-cache-parsedfile-coverage.test.ts index dabca8001..36e9140ce 100644 --- a/gitnexus/test/unit/parse-impl-warm-cache-parsedfile-coverage.test.ts +++ b/gitnexus/test/unit/parse-impl-warm-cache-parsedfile-coverage.test.ts @@ -26,12 +26,28 @@ * shards are absent; and a mixed-mode run (one file changed) hits the * unchanged chunk while re-parsing the changed one. */ -import { afterEach, beforeEach, describe, expect, it } from 'vitest'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import fs from 'node:fs'; import os from 'node:os'; import path from 'node:path'; import { pathToFileURL } from 'node:url'; +// Partial mock: lets one test make prepareDurableParsedFileChunk fail without +// touching the worker-side persist path (which shares the same directory). +const prepareOverride = vi.hoisted(() => ({ + impl: undefined as undefined | (() => Promise), +})); +vi.mock('../../src/storage/parsedfile-store.js', async (importOriginal) => { + const real = await importOriginal(); + return { + ...real, + prepareDurableParsedFileChunk: (durableDir: string, chunkHash: string) => + prepareOverride.impl + ? prepareOverride.impl() + : real.prepareDurableParsedFileChunk(durableDir, chunkHash), + }; +}); + import { createKnowledgeGraph } from '../../src/core/graph/graph.js'; import { runChunkedParseAndResolve } from '../../src/core/ingestion/pipeline-phases/parse-impl.js'; import { @@ -315,6 +331,16 @@ describe('parse-impl warm-cache ParsedFile coverage (#2038)', () => { expect(cache.usedKeys.has(chunkHash)).toBe(true); }); + it('a failing durable-generation reset degrades instead of failing the analyze', async () => { + const f = writeFile('src/degrade.ts', 'export function degrade() { return 1; }\n'); + prepareOverride.impl = () => Promise.reject(new Error('EACCES: simulated cache failure')); + try { + await expect(run(newCache(), [f])).resolves.toBeUndefined(); + } finally { + prepareOverride.impl = undefined; + } + }); + it('a repeated cache miss replaces the durable chunk generation', async () => { const f = writeFile('src/repeated.ts', 'export function repeated() { return 1; }\n'); const chunkHash = computeChunkHash([ From e20e326290ae2affd068a9391f4f527835a222ef Mon Sep 17 00:00:00 2001 From: Gergo Magyar Date: Thu, 16 Jul 2026 09:58:14 +0000 Subject: [PATCH 7/9] fix(cli): fail every tool command loudly on backend error payloads Moves the #2469 guard from cypherCommand into output() so all seven tool commands that print backend results share the exit semantics. Adds query and context regression cases. Co-Authored-By: Claude Fable 5 --- gitnexus/src/cli/tool.ts | 22 +++++++++++++--------- gitnexus/test/unit/tool-direct-cli.test.ts | 20 ++++++++++++++++++++ 2 files changed, 33 insertions(+), 9 deletions(-) diff --git a/gitnexus/src/cli/tool.ts b/gitnexus/src/cli/tool.ts index 46adaf990..060571431 100644 --- a/gitnexus/src/cli/tool.ts +++ b/gitnexus/src/cli/tool.ts @@ -56,6 +56,19 @@ function output(data: any): void { // Fallback: stderr (previous behavior, works on all platforms) process.stderr.write(text + '\n'); } + // Backend failures come back as `{ error }` payloads rather than throws + // (#2469). Every tool command routes its result through here, so this is + // the one place that keeps scripted callers honest: print the payload, + // then exit non-zero. + if ( + data && + typeof data === 'object' && + 'error' in data && + typeof data.error === 'string' && + data.error.trim().length > 0 + ) { + process.exitCode = 1; + } } /** @@ -301,15 +314,6 @@ export async function cypherCommand( } } output(result); - if ( - result && - typeof result === 'object' && - 'error' in result && - typeof result.error === 'string' && - result.error.trim().length > 0 - ) { - process.exitCode = 1; - } } export async function detectChangesCommand(options?: { diff --git a/gitnexus/test/unit/tool-direct-cli.test.ts b/gitnexus/test/unit/tool-direct-cli.test.ts index 2230a675e..66b031231 100644 --- a/gitnexus/test/unit/tool-direct-cli.test.ts +++ b/gitnexus/test/unit/tool-direct-cli.test.ts @@ -104,6 +104,26 @@ describe('direct CLI tool commands', () => { expect(process.exitCode).toBeUndefined(); }); + it('fails closed when query returns a backend error payload', async () => { + callToolMock.mockResolvedValue({ error: 'Repository "missing" not found.' }); + const { queryCommand } = await import('../../src/cli/tool.js'); + + await queryCommand('auth flow'); + + expect(writeSyncMock).toHaveBeenCalledWith(1, expect.stringContaining('not found')); + expect(process.exitCode).toBe(1); + }); + + it('fails closed when context returns a backend error payload', async () => { + callToolMock.mockResolvedValue({ error: 'Symbol not found: nope' }); + const { contextCommand } = await import('../../src/cli/tool.js'); + + await contextCommand('nope'); + + expect(writeSyncMock).toHaveBeenCalledWith(1, expect.stringContaining('Symbol not found')); + expect(process.exitCode).toBe(1); + }); + it('dispatches detect_changes with CLI-shaped arguments', async () => { callToolMock.mockResolvedValue({ summary: { From e814c5a10d63e1ebe58d3951f7ca2244ce4c1d2e Mon Sep 17 00:00:00 2001 From: Gergo Magyar Date: Thu, 16 Jul 2026 10:00:26 +0000 Subject: [PATCH 8/9] fix(embeddings): include File rows in the incremental delete sweeps The zero-symbol File fallback from #2455 writes File embedding rows, but the filePath-scoped delete sweeps joined through EMBEDDABLE_LABELS only. Docs repos accumulated duplicate rows on re-analyze and deleted files left orphans. Free for code repos: no File rows exist to match. Co-Authored-By: Claude Fable 5 --- gitnexus/src/core/lbug/lbug-adapter.ts | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/gitnexus/src/core/lbug/lbug-adapter.ts b/gitnexus/src/core/lbug/lbug-adapter.ts index 72389e915..625167a0c 100644 --- a/gitnexus/src/core/lbug/lbug-adapter.ts +++ b/gitnexus/src/core/lbug/lbug-adapter.ts @@ -2124,16 +2124,18 @@ export const isLbugReady = (): boolean => conn !== null && db !== null; /** * Multi-label alternation over exactly the labels that can own embedding - * rows (embedding-pipeline.ts queries EMBEDDABLE_LABELS and nothing else), - * reserved keywords backtick-escaped via {@link escapeTableName}. Probed on - * @ladybugdb/core 0.18.0 (this shipping review, FIX 4): the full 19-label + * rows: EMBEDDABLE_LABELS plus File, which embedding-pipeline.ts embeds as + * the zero-symbol fallback for text-only repositories (#2454). Reserved + * keywords are backtick-escaped via {@link escapeTableName}. Probed on + * @ladybugdb/core 0.18.0 (this shipping review, FIX 4): the full multi-label * alternation parses, executes, and deletes exactly the joined rows — * replacing the unlabeled `MATCH (n)` that scanned EVERY node table per * chunk (BasicBlock-dominated under `--pdg`) when only embeddable labels - * can match an embedding row. + * can match an embedding row. Including File is free for code repositories: + * they never hold File embedding rows, so the extra label joins nothing. */ const embeddableLabelMatch = (): string => - EMBEDDABLE_LABELS.map((l) => escapeTableName(l)).join('|'); + ['File', ...EMBEDDABLE_LABELS].map((l) => escapeTableName(l)).join('|'); // LADYBUGDB-CONTRACT: matches @ladybugdb/core ^0.18.0 native binder text, // probe-recorded: `Binder exception: Table CodeEmbedding does not exist.` From 36d25b5a7072ea008671977344ab447ae578d898 Mon Sep 17 00:00:00 2001 From: Gergo Magyar Date: Thu, 16 Jul 2026 10:11:32 +0000 Subject: [PATCH 9/9] test(cli): pin the non-zero exit for not-found context payloads The skip-git ignore test asserted the error payload while relying on exit 0; since the output() guard an error payload also exits 1, so the test now captures the payload from the exec failure and pins both. Co-Authored-By: Claude Fable 5 --- gitnexus/test/unit/skip-git-cli.test.ts | 27 +++++++++++++++++-------- 1 file changed, 19 insertions(+), 8 deletions(-) diff --git a/gitnexus/test/unit/skip-git-cli.test.ts b/gitnexus/test/unit/skip-git-cli.test.ts index 0be164ef6..0b3bed936 100644 --- a/gitnexus/test/unit/skip-git-cli.test.ts +++ b/gitnexus/test/unit/skip-git-cli.test.ts @@ -158,14 +158,25 @@ describe('--skip-git CLI flag', () => { expect(keepContext).toContain('"status": "found"'); expect(keepContext).toContain('"filePath": "src/keep.ts"'); - const leakedContext = execSync( - `node "${cliPath}" context leaked --repo "${path.basename(tmpDir)}"`, - { - encoding: 'utf8', - timeout: 60000, - env, - }, - ); + // Since #2470 a backend error payload also exits non-zero, so capture + // the payload from the exec failure instead of expecting exit 0. + let leakedContext = ''; + let leakedStatus = 0; + try { + leakedContext = execSync( + `node "${cliPath}" context leaked --repo "${path.basename(tmpDir)}"`, + { + encoding: 'utf8', + timeout: 60000, + env, + }, + ); + } catch (err: unknown) { + const execErr = err as { status?: number; stdout?: string | Buffer }; + leakedStatus = execErr.status ?? 0; + leakedContext = String(execErr.stdout ?? ''); + } + expect(leakedStatus).toBe(1); expect(leakedContext).toContain(`"error": "Symbol 'leaked' not found"`); } finally { fs.rmSync(tmpDir, { recursive: true, force: true });