mirror of
https://github.com/abhigyanpatwari/GitNexus.git
synced 2026-08-28 05:25:25 +00:00
Merge pull request #2512 from abhigyanpatwari/merge/eva-fixes-2
fix: land cache, CLI, and embeddings series (#2476 #2470 #2455 #2468)
This commit is contained in:
Gergő Magyar
GitHub
commit
c4fb511a0a
25 changed files with 1354 additions and 50 deletions
|
|
@ -33,6 +33,8 @@ export interface ResilientFetchOptions {
|
|||
breakerOptions?: CircuitBreakerOptions;
|
||||
/** Tuning knobs for the retry helper. */
|
||||
retry?: Partial<Pick<RetryOptions, 'maxAttempts' | 'baseDelayMs' | 'capDelayMs'>> & {
|
||||
/** 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':
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -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<Float32Array> => {
|
||||
export const embedText = async (
|
||||
text: string,
|
||||
options: EmbeddingRequestOptions = {},
|
||||
): Promise<Float32Array> => {
|
||||
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<Float32Array> => {
|
|||
* @param texts - Array of texts to embed
|
||||
* @returns Array of Float32Array embedding vectors
|
||||
*/
|
||||
export const embedBatch = async (texts: string[]): Promise<Float32Array[]> => {
|
||||
export const embedBatch = async (
|
||||
texts: string[],
|
||||
options: EmbeddingRequestOptions = {},
|
||||
): Promise<Float32Array[]> => {
|
||||
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<Float32Array[]> => {
|
|||
pooling: 'mean',
|
||||
normalize: true,
|
||||
});
|
||||
options.signal?.throwIfAborted();
|
||||
|
||||
// Result shape is [batch_size, dimensions]
|
||||
// Need to split into individual vectors
|
||||
|
|
|
|||
32
gitnexus/src/core/embeddings/embedding-identity.ts
Normal file
32
gitnexus/src/core/embeddings/embedding-identity.ts
Normal file
|
|
@ -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,
|
||||
};
|
||||
}
|
||||
|
|
@ -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<any[]>,
|
||||
): Promise<EmbeddableNode[]> => {
|
||||
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 [];
|
||||
}
|
||||
};
|
||||
|
||||
/**
|
||||
|
|
@ -262,6 +305,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<string>;
|
||||
onCheckpointWindowStart?: (window: EmbeddingPipelineCheckpointWindow) => Promise<void>;
|
||||
onCheckpoint?: (checkpoint: EmbeddingPipelineCheckpoint) => Promise<void>;
|
||||
}
|
||||
|
||||
/**
|
||||
* DELETE stale embedding rows for the given nodeIds so they can be re-inserted.
|
||||
*
|
||||
|
|
@ -320,12 +381,20 @@ export const runEmbeddingPipeline = async (
|
|||
config: Partial<EmbeddingConfig> = {},
|
||||
skipNodeIds?: Set<string>,
|
||||
existingEmbeddings?: Map<string, string>,
|
||||
pipelineOptions: EmbeddingPipelineOptions = {},
|
||||
): Promise<EmbeddingPipelineResult> => {
|
||||
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 +415,7 @@ export const runEmbeddingPipeline = async (
|
|||
modelDownloadPercent: downloadPercent,
|
||||
});
|
||||
}, finalConfig);
|
||||
throwIfCancelled();
|
||||
}
|
||||
|
||||
onProgress({
|
||||
|
|
@ -360,6 +430,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 +441,20 @@ export const runEmbeddingPipeline = async (
|
|||
// than all up front — see U6 / KTD7. `staleNodeIds` is consulted inside the
|
||||
// batch loop; it stays empty in full (non-incremental) mode so no deletes fire.
|
||||
const staleNodeIds = new Set<string>();
|
||||
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 +471,14 @@ export const runEmbeddingPipeline = async (
|
|||
}
|
||||
}
|
||||
|
||||
if (forceReembedNodeIds && forceReembedNodeIds.size > 0) {
|
||||
const removedPendingNodeIds = [...forceReembedNodeIds].filter(
|
||||
(nodeId) => !embeddableNodeIds.has(nodeId),
|
||||
);
|
||||
await deleteStaleEmbeddingRows(executeWithReusedStatement, removedPendingNodeIds);
|
||||
throwIfCancelled();
|
||||
}
|
||||
|
||||
const totalNodes = nodes.length;
|
||||
|
||||
if (isDev) {
|
||||
|
|
@ -402,6 +486,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 +510,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 +527,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 +621,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 +631,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 +646,7 @@ export const runEmbeddingPipeline = async (
|
|||
}));
|
||||
|
||||
await batchInsertEmbeddings(executeWithReusedStatement, dbUpdates);
|
||||
throwIfCancelled();
|
||||
}
|
||||
|
||||
processedNodes += batch.length;
|
||||
|
|
@ -558,9 +661,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,
|
||||
|
|
|
|||
|
|
@ -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<void> = 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<void> => {
|
||||
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<void> => {
|
||||
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<EmbeddingItem[]> => {
|
||||
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<Float32Array[]> => {
|
||||
export const httpEmbed = async (
|
||||
texts: string[],
|
||||
requestOptions: EmbeddingRequestOptions = {},
|
||||
): Promise<Float32Array[]> => {
|
||||
if (texts.length === 0) return [];
|
||||
|
||||
const config = readConfig();
|
||||
|
|
@ -309,6 +435,10 @@ export const httpEmbed = async (texts: string[]): Promise<Float32Array[]> => {
|
|||
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<Float32Array[]> => {
|
|||
* @param text - Query text to embed
|
||||
* @returns Embedding vector as number array
|
||||
*/
|
||||
export const httpEmbedQuery = async (text: string): Promise<number[]> => {
|
||||
export const httpEmbedQuery = async (
|
||||
text: string,
|
||||
requestOptions: EmbeddingRequestOptions = {},
|
||||
): Promise<number[]> => {
|
||||
const config = readConfig();
|
||||
if (!config) throw new Error('HTTP embedding not configured');
|
||||
|
||||
|
|
@ -359,6 +492,10 @@ export const httpEmbedQuery = async (text: string): Promise<number[]> => {
|
|||
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)})`);
|
||||
|
|
|
|||
|
|
@ -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,19 @@ 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) {
|
||||
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;
|
||||
// Parse phase covers 20-70 (M2). Deferred extraction handles 70-95.
|
||||
|
|
|
|||
|
|
@ -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.`
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
@ -760,6 +761,44 @@ export async function runFullAnalysis(
|
|||
}
|
||||
}
|
||||
|
||||
let resumeEmbeddingCheckpoint = false;
|
||||
let pendingEmbeddingNodeIds = new Set<string>();
|
||||
let embeddingIdentityForRun: EmbeddingIdentity | undefined;
|
||||
if (existingMeta?.embeddingCheckpoint) {
|
||||
if (options.dropEmbeddings) {
|
||||
log('Discarding the interrupted embedding checkpoint (--drop-embeddings).');
|
||||
options = { ...options, force: true };
|
||||
} else {
|
||||
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
|
||||
) {
|
||||
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 +937,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 +1075,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 +1781,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 +1847,11 @@ export async function runFullAnalysis(
|
|||
httpMode ? 'Connecting to embedding endpoint...' : 'Loading embedding model...',
|
||||
);
|
||||
const { runEmbeddingPipeline } = await import('./embeddings/embedding-pipeline.js');
|
||||
if (!embeddingIdentityForRun) {
|
||||
const { resolveEmbeddingIdentity } = await import('./embeddings/embedding-identity.js');
|
||||
embeddingIdentityForRun = resolveEmbeddingIdentity();
|
||||
}
|
||||
const embeddingIdentity = embeddingIdentityForRun;
|
||||
// Build a Map<nodeId, contentHash> from cached embeddings for incremental mode
|
||||
let existingEmbeddings: Map<string, string> | undefined;
|
||||
if (cachedEmbeddingNodeIds.size > 0) {
|
||||
|
|
@ -1810,6 +1861,49 @@ export async function runFullAnalysis(
|
|||
}
|
||||
}
|
||||
|
||||
const saveEmbeddingCheckpoint = async (
|
||||
checkpoint: {
|
||||
nodesProcessed: number;
|
||||
totalNodes: number;
|
||||
chunksProcessed: number;
|
||||
},
|
||||
pendingNodeIds: string[],
|
||||
embeddings: number | undefined,
|
||||
): Promise<void> => {
|
||||
const fileHashes: Record<string, string> = {};
|
||||
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,
|
||||
provider: embeddingIdentity.provider,
|
||||
pendingNodeIds,
|
||||
},
|
||||
pdg: resolvePdgConfig(options),
|
||||
});
|
||||
};
|
||||
|
||||
const embeddingResult = await runEmbeddingPipeline(
|
||||
executeQuery,
|
||||
executeWithReusedStatement,
|
||||
|
|
@ -1826,6 +1920,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 +2061,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
|
||||
|
|
|
|||
|
|
@ -40,6 +40,7 @@ const JOB_TIMEOUT_MS = 30 * 60 * 1000; // 30 minutes
|
|||
export class JobManager {
|
||||
private jobs = new Map<string, AnalyzeJob>();
|
||||
private children = new Map<string, ChildProcess>();
|
||||
private abortControllers = new Map<string, AbortController>();
|
||||
private timeouts = new Map<string, ReturnType<typeof setTimeout>>();
|
||||
private emitter = new EventEmitter();
|
||||
private cleanupTimer: ReturnType<typeof setInterval>;
|
||||
|
|
@ -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);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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 { 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 ||
|
||||
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<void> => {
|
||||
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);
|
||||
}
|
||||
})();
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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<void> => {
|
||||
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<string>();
|
||||
|
|
|
|||
|
|
@ -197,6 +197,30 @@ 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;
|
||||
/** `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,
|
||||
* 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
|
||||
|
|
|
|||
|
|
@ -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' });
|
||||
|
|
|
|||
|
|
@ -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');
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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();
|
||||
|
||||
|
|
@ -549,6 +621,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();
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
@ -28,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);
|
||||
|
|
@ -339,9 +364,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 +604,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', () => {
|
||||
|
|
|
|||
|
|
@ -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 () => {});
|
||||
|
|
|
|||
|
|
@ -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<void>),
|
||||
}));
|
||||
vi.mock('../../src/storage/parsedfile-store.js', async (importOriginal) => {
|
||||
const real = await importOriginal<typeof import('../../src/storage/parsedfile-store.js')>();
|
||||
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 {
|
||||
|
|
@ -41,6 +57,7 @@ import {
|
|||
} from '../../src/storage/parse-cache.js';
|
||||
import {
|
||||
getDurableParsedFileDir,
|
||||
prepareDurableParsedFileChunk,
|
||||
persistDurableParsedFileShardSync,
|
||||
restoreDurableParsedFileShard,
|
||||
loadParsedFilesForPaths,
|
||||
|
|
@ -100,6 +117,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 +331,37 @@ 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([
|
||||
{
|
||||
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();
|
||||
|
|
|
|||
|
|
@ -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,172 @@ 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');
|
||||
const { resolveEmbeddingIdentity } =
|
||||
await import('../../src/core/embeddings/embedding-identity.js');
|
||||
const embeddingIdentity = resolveEmbeddingIdentity();
|
||||
await saveMeta(storagePath, {
|
||||
...completed,
|
||||
embeddingCheckpoint: {
|
||||
at: new Date().toISOString(),
|
||||
nodesProcessed: 1,
|
||||
totalNodes: 1,
|
||||
chunksProcessed: 1,
|
||||
model: 'test-model',
|
||||
dimensions: 384,
|
||||
provider: embeddingIdentity.provider,
|
||||
},
|
||||
} 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,
|
||||
provider: embeddingIdentity.provider,
|
||||
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: '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: {
|
||||
at: new Date().toISOString(),
|
||||
nodesProcessed: 1,
|
||||
totalNodes: 2,
|
||||
chunksProcessed: 1,
|
||||
model: 'different-model',
|
||||
dimensions: 384,
|
||||
provider: embeddingIdentity.provider,
|
||||
},
|
||||
});
|
||||
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-');
|
||||
|
|
|
|||
|
|
@ -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 });
|
||||
|
|
|
|||
|
|
@ -81,6 +81,49 @@ 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('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: {
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue