diff --git a/README.md b/README.md index 2f55daa2a..829f007a4 100644 --- a/README.md +++ b/README.md @@ -583,6 +583,7 @@ Most `analyze` knobs are also CLI flags (`--workers`, `--worker-timeout`, `--max | `GITNEXUS_WORKER_POOL_SIZE` | `cores - 1`, capped at 16 | Parse worker pool size (must be ≥ 1). Equivalent to `--workers `. The worker pool is the sole parse path — there is no sequential parser, so `0` is rejected with an actionable error (the pool self-heals via quarantine + respawn). | Constrained containers (cgroup CPU limits) or CI runners with explicit quotas. To narrow down a worker crash set `1` for a single-worker pool — not `0`. | | `GITNEXUS_PARSE_CHUNK_CONCURRENCY` | `2` | Number of chunks whose file contents may be read into memory in parallel while the pool dispatches the current chunk. Worker dispatch itself stays serial. | Repos large enough to chunk (multi-MB total source) where disk I/O is a measurable fraction of analyze wall-clock. | | `GITNEXUS_VERBOSE` | unset | When `1`, enables verbose ingestion logs (skipped-file warnings, per-chunk throughput, parse-cache stats). Equivalent to `--verbose`. | Debugging an analyze that "completed" but seems to have missed files; tuning `--workers` / chunk concurrency against observable throughput. | +| `GITNEXUS_EMBEDDING_RETRY_TIMEOUTS` | unset | When `1`, per-attempt HTTP embedding timeouts (`TimeoutError` on fetch or body read) go through the bounded `GITNEXUS_EMBEDDING_MAX_ATTEMPTS` retry loop instead of failing the job. Default stays off so cloud/default timeouts remain terminal. | Local accelerators that drop a device lock when the client disconnects and succeed on the next request (observed with FastFlowLM on Ryzen AI). | | `GITNEXUS_ANALYZER_IDENTITY_IN_PROCESS_GUARDS` | unset | When truthy (`1`/`true`/`yes`), forces in-process cache-guard validation once a batch has ≥128 requests. In-process mode also auto-selects when `packageRoot`/`buildRoot` fail `W_OK` with `EACCES`/`EROFS`. Otherwise those large batches use a Node subprocess probe. Batches under 128 always stay in-process. | Trusted or read-only installs where two identity subprocess spawns per analyze dominate wall time; leave unset to keep the default isolation path on writable trees. | | `GITNEXUS_RESOLVE_DEF_GRAPH_ID_MEMO` | on (unset) | Memoizes `resolveDefGraphId` per `nodeLookup` instance (WeakMap). Enabled by default. Set to `0`/`false`/`off`/`no` to disable and recompute on every call (debug / bisect memo bugs). | Suspecting stale graph-id resolution after a lookup rebuild, or comparing memo vs uncached cost on a large index. | | `GITNEXUS_AUTH_TOKEN` | unset | Bearer token required when `eval-server` binds beyond loopback. May also be read from `.env.local` or `.env`; shell values take precedence. | Exposing the evaluation HTTP tools to a container, VM, or LAN. | diff --git a/gitnexus/src/cli/embeddings-sync.ts b/gitnexus/src/cli/embeddings-sync.ts new file mode 100644 index 000000000..a54831c76 --- /dev/null +++ b/gitnexus/src/cli/embeddings-sync.ts @@ -0,0 +1,170 @@ +import { lstat } from 'node:fs/promises'; +import path from 'node:path'; +import { cliInfo } from './cli-message.js'; +import { getGitRoot } from '../storage/git.js'; +import { acquireIndexLock } from '../storage/index-lock.js'; +import { getStoragePaths, loadMeta, saveMeta } from '../storage/repo-manager.js'; +import { + closeLbug, + executeQuery, + executeWithReusedStatement, + fetchExistingEmbeddingHashes, + initLbug, +} from '../core/lbug/lbug-adapter.js'; +import { runEmbeddingPipeline } from '../core/embeddings/embedding-pipeline.js'; +import { resolveEmbeddingIdentity } from '../core/embeddings/embedding-identity.js'; +import { + checkpointKind, + decideEmbeddingResume, + mintInterruptedCheckpoint, + mintPartialCheckpoint, + mintUnverifiedCountCheckpoint, + type EmbeddingCheckpoint, + type EmbeddingCheckpointProgress, +} from '../core/embedding-checkpoint.js'; +import { + measurePersistedEmbeddingCount, + persistedEmbeddingCountOrUndefined, +} from '../core/embedding-count.js'; + +/** Add missing embeddings directly to a healthy index, checkpointing periodically. */ +export const embeddingsSyncCommand = async (inputPath?: string): Promise => { + const repoPath = inputPath ? path.resolve(inputPath) : getGitRoot(process.cwd()); + if (!repoPath) throw new Error('Not inside a git repository. Pass a repository path.'); + + const { lbugPath, metaPath } = getStoragePaths(repoPath); + const metaDir = path.dirname(metaPath); + const lock = await acquireIndexLock(metaDir); + try { + const meta = await loadMeta(metaDir); + if (!meta) + throw new Error(`No GitNexus index found for ${repoPath}. Run gitnexus analyze first.`); + if (meta.incrementalInProgress) { + throw new Error('The structural index is incomplete. Run gitnexus analyze --force first.'); + } + + let lbugStat; + try { + lbugStat = await lstat(lbugPath); + } catch { + throw new Error( + `The LadybugDB graph store at ${lbugPath} is missing. Run gitnexus analyze first.`, + ); + } + if (!lbugStat.isFile()) { + throw new Error( + `The LadybugDB graph store at ${lbugPath} is not a usable database file. Run gitnexus analyze first.`, + ); + } + + const identity = resolveEmbeddingIdentity(); + let forceReembedNodeIds: ReadonlySet | undefined; + let resumedFrom: EmbeddingCheckpoint | undefined; + if (meta.embeddingCheckpoint) { + const checkpoint = meta.embeddingCheckpoint; + const decision = decideEmbeddingResume(checkpoint, identity); + if (decision.action === 'abort') throw new Error(decision.error); + const identityDiffers = + checkpoint.provider !== identity.provider || + checkpoint.model !== identity.model || + checkpoint.dimensions !== identity.dimensions; + // `abandon` on a non-interrupted foreign identity drops the pending set + // only. Existing rows stay; sync would then embed the holes under the new + // identity and mix vector spaces. Fail closed — rebuild via analyze. + if (identityDiffers && checkpointKind(checkpoint) !== 'unverified-count') { + throw new Error( + `Cannot sync embeddings: the index checkpoint was written by ${checkpoint.model} ` + + `(${checkpoint.provider}) at ${checkpoint.dimensions} dimensions, but this run ` + + `resolves ${identity.model} (${identity.provider}) at ${identity.dimensions}. ` + + 'Run `gitnexus analyze --embeddings --force` to rebuild under the new identity.', + ); + } + cliInfo(decision.log); + if (decision.action === 'resume') { + forceReembedNodeIds = decision.pendingNodeIds; + resumedFrom = decision.resumedFrom; + } + } + + await initLbug(lbugPath); + try { + const existing = await fetchExistingEmbeddingHashes(executeQuery); + let lastPercent = -1; + + const countEmbeddings = async (): Promise => + persistedEmbeddingCountOrUndefined(await measurePersistedEmbeddingCount(executeQuery)); + const saveCheckpoint = async ( + checkpoint: EmbeddingCheckpointProgress, + pendingNodeIds: string[], + embeddings?: number, + ): Promise => { + const latest = (await loadMeta(metaDir)) ?? meta; + await saveMeta(metaDir, { + ...latest, + ...(embeddings === undefined ? {} : { stats: { ...latest.stats, embeddings } }), + embeddingCheckpoint: mintInterruptedCheckpoint(identity, checkpoint, pendingNodeIds), + }); + }; + + cliInfo(`Embedding ${repoPath}`); + cliInfo(`Checkpointed nodes already present: ${existing?.size ?? 0}`); + + const result = await runEmbeddingPipeline( + executeQuery, + executeWithReusedStatement, + (progress) => { + const percent = Math.floor(progress.percent); + if (percent !== lastPercent && (percent % 5 === 0 || percent === 100)) { + lastPercent = percent; + cliInfo( + ` ${percent}% — ${progress.nodesProcessed ?? 0}/${progress.totalNodes ?? '?'} nodes`, + ); + } + }, + {}, + undefined, + existing && existing.size ? existing : undefined, + { + forceReembedNodeIds, + onCheckpointWindowStart: async ({ nodeIds, ...checkpoint }) => { + await saveCheckpoint(checkpoint, nodeIds); + }, + onCheckpoint: async (checkpoint) => { + await saveCheckpoint(checkpoint, [], await countEmbeddings()); + }, + }, + ); + + const embeddings = await countEmbeddings(); + const latest = (await loadMeta(metaDir)) ?? meta; + if (embeddings === undefined) { + // Keep last-known stats.embeddings. An interrupted window marker would + // fail the identity gate on the next run even though this run finished; + // unverified-count is the recovery kind that forces a recount (#2790). + await saveMeta(metaDir, { + ...latest, + embeddingCheckpoint: result.failedNodeIds.length + ? mintPartialCheckpoint(identity, result, resumedFrom) + : mintUnverifiedCountCheckpoint(identity, { + nodesProcessed: result.nodesProcessed, + totalNodes: result.nodesProcessed, + chunksProcessed: result.chunksProcessed, + }), + }); + throw new Error('Could not verify persisted embedding count.'); + } + await saveMeta(metaDir, { + ...latest, + stats: { ...latest.stats, embeddings }, + embeddingCheckpoint: result.failedNodeIds.length + ? mintPartialCheckpoint(identity, result, resumedFrom) + : undefined, + }); + cliInfo(`Embeddings ready: ${embeddings}`); + } finally { + await closeLbug().catch(() => {}); + } + } finally { + lock.release(); + } +}; diff --git a/gitnexus/src/cli/i18n/en.ts b/gitnexus/src/cli/i18n/en.ts index 3585a6648..8fe3025fd 100644 --- a/gitnexus/src/cli/i18n/en.ts +++ b/gitnexus/src/cli/i18n/en.ts @@ -168,6 +168,8 @@ export const en = { 'error.watch.ambiguous': '`gitnexus watch` is ambiguous.\n Local working-tree incremental index: gitnexus analyze --watch\n Scheduled remote clone/pull + analyze: gitnexus auto-sync start\n', 'help.command.analyze.description': 'Index a repository (full analysis)', + 'help.command.embeddings.sync.description': + 'Add missing embeddings to an existing index, checkpointing periodically for safe resume', 'help.command.index.description': 'Register an existing .gitnexus/ folder into the global registry (no re-analysis needed)', 'help.command.serve.description': 'Start local HTTP server for web UI connection', @@ -356,5 +358,5 @@ export const en = { 'help.identityCache.environment': '\nAnalyzer identity cache:\n GITNEXUS_ANALYZER_IDENTITY_CACHE_DIR=/absolute/protected/dir\n Operator-trusted persistent cache for warm cross-process status. The directory must pre-exist, be outside the GitNexus package/build roots, and contain no symlink or junction components. Defaults remain fail-closed on platforms without POSIX ownership APIs.', 'help.analyze.environment': - '\nEnvironment variables:\n GITNEXUS_NO_GITIGNORE=1 Skip .gitignore parsing (still reads .gitnexusignore)\n GITNEXUS_MAX_FILE_SIZE=N Override large-file skip threshold (KB). Default 512, max 32768.\n GITNEXUS_ANALYZER_IDENTITY_CACHE_DIR=/absolute/protected/dir Operator-trusted persistent analyzer identity cache; must pre-exist, be outside package/build roots, and contain no symlink/junction components.\n GITNEXUS_WORKER_SUB_BATCH_TIMEOUT_MS=N Worker idle timeout in milliseconds. Default 30000.\n GITNEXUS_WAL_CHECKPOINT_THRESHOLD=N LadybugDB WAL auto-checkpoint threshold in bytes (default 67108864 = 64 MiB; -1 keeps Ladybug stock ~16 MiB).\n GITNEXUS_WORKER_SUB_BATCH_MAX_BYTES=N Worker job byte budget. Default 8388608.\n GITNEXUS_WORKER_POOL_SIZE=N Parse worker count override. Default cores-1 capped at 16.\n GITNEXUS_PARSE_CHUNK_CONCURRENCY=N Concurrent in-flight parse chunks. Default 2.\n GITNEXUS_WORKER_MAX_RESPAWNS_PER_SLOT=N Max replacement spawns per slot before drop. Default 3.\n GITNEXUS_WORKER_MAX_CUMULATIVE_TIMEOUT_MS=N Total retry wall-time per job. Default 5x sub-batch timeout.\n GITNEXUS_WORKER_CONSECUTIVE_FAILURE_THRESHOLD=N Per-slot deaths to trip circuit breaker. Default max(3, poolSize).\n GITNEXUS_WORKER_SHUTDOWN_DRAIN_MS=N Max wait at pool shutdown for a retired worker still inside native code (terminated at its next safe point instead of aborting the process). Default 30000.\n GITNEXUS_CPP_CAPTURE_BUDGET_MS=N Per-file wall-clock budget for C++ capture extraction; on breach the file keeps partial captures with a warning. Default 20000.\n GITNEXUS_EMBEDDING_THREADS=N Limit local ONNX CPU threads for --embeddings.\n GITNEXUS_SEMANTIC_EXACT_SCAN_LIMIT=N Max embedding chunks for exact-scan fallback. Default 10000.\n GITNEXUS_VECTOR_MAX_DISTANCE=N Max accepted semantic/vector cosine distance (0 < N <= 2; higher values clamp to 2). Default 0.6 for MCP, 0.5 elsewhere.\n\nFlags override the corresponding env vars when both are provided.\n\nTip: `.gitnexusignore` supports `.gitignore`-style negation. Add e.g.\n `!__tests__/` to index a directory that is auto-filtered by default (#771).', + '\nEnvironment variables:\n GITNEXUS_NO_GITIGNORE=1 Skip .gitignore parsing (still reads .gitnexusignore)\n GITNEXUS_MAX_FILE_SIZE=N Override large-file skip threshold (KB). Default 512, max 32768.\n GITNEXUS_ANALYZER_IDENTITY_CACHE_DIR=/absolute/protected/dir Operator-trusted persistent analyzer identity cache; must pre-exist, be outside package/build roots, and contain no symlink/junction components.\n GITNEXUS_WORKER_SUB_BATCH_TIMEOUT_MS=N Worker idle timeout in milliseconds. Default 30000.\n GITNEXUS_WAL_CHECKPOINT_THRESHOLD=N LadybugDB WAL auto-checkpoint threshold in bytes (default 67108864 = 64 MiB; -1 keeps Ladybug stock ~16 MiB).\n GITNEXUS_WORKER_SUB_BATCH_MAX_BYTES=N Worker job byte budget. Default 8388608.\n GITNEXUS_WORKER_POOL_SIZE=N Parse worker count override. Default cores-1 capped at 16.\n GITNEXUS_PARSE_CHUNK_CONCURRENCY=N Concurrent in-flight parse chunks. Default 2.\n GITNEXUS_WORKER_MAX_RESPAWNS_PER_SLOT=N Max replacement spawns per slot before drop. Default 3.\n GITNEXUS_WORKER_MAX_CUMULATIVE_TIMEOUT_MS=N Total retry wall-time per job. Default 5x sub-batch timeout.\n GITNEXUS_WORKER_CONSECUTIVE_FAILURE_THRESHOLD=N Per-slot deaths to trip circuit breaker. Default max(3, poolSize).\n GITNEXUS_WORKER_SHUTDOWN_DRAIN_MS=N Max wait at pool shutdown for a retired worker still inside native code (terminated at its next safe point instead of aborting the process). Default 30000.\n GITNEXUS_CPP_CAPTURE_BUDGET_MS=N Per-file wall-clock budget for C++ capture extraction; on breach the file keeps partial captures with a warning. Default 20000.\n GITNEXUS_EMBEDDING_THREADS=N Limit local ONNX CPU threads for --embeddings.\n GITNEXUS_EMBEDDING_RETRY_TIMEOUTS=1 Retry per-attempt HTTP embedding timeouts through GITNEXUS_EMBEDDING_MAX_ATTEMPTS (default off; timeouts stay terminal).\n GITNEXUS_SEMANTIC_EXACT_SCAN_LIMIT=N Max embedding chunks for exact-scan fallback. Default 10000.\n GITNEXUS_VECTOR_MAX_DISTANCE=N Max accepted semantic/vector cosine distance (0 < N <= 2; higher values clamp to 2). Default 0.6 for MCP, 0.5 elsewhere.\n\nFlags override the corresponding env vars when both are provided.\n\nTip: `.gitnexusignore` supports `.gitignore`-style negation. Add e.g.\n `!__tests__/` to index a directory that is auto-filtered by default (#771).', } as const; diff --git a/gitnexus/src/cli/i18n/zh-CN.ts b/gitnexus/src/cli/i18n/zh-CN.ts index 2f84f4b97..695acd72a 100644 --- a/gitnexus/src/cli/i18n/zh-CN.ts +++ b/gitnexus/src/cli/i18n/zh-CN.ts @@ -164,6 +164,8 @@ export const zhCN = { 'error.watch.ambiguous': '`gitnexus watch` 含义不明确。\n 本地工作区增量索引：gitnexus analyze --watch\n 定时远程 clone/pull 并分析：gitnexus auto-sync start\n', 'help.command.analyze.description': '索引仓库（完整分析）', + 'help.command.embeddings.sync.description': + '向现有索引添加缺失的嵌入，并定期保存检查点以安全续跑', 'help.command.index.description': '将现有 .gitnexus/ 文件夹注册到全局注册表（无需重新分析）', 'help.command.serve.description': '启动供 Web UI 连接的本地 HTTP 服务器', 'help.command.mcp.description': @@ -329,5 +331,5 @@ export const zhCN = { 'help.identityCache.environment': '\n分析器身份缓存：\n GITNEXUS_ANALYZER_IDENTITY_CACHE_DIR=/absolute/protected/dir\n 由操作员明确信任的持久缓存，用于跨进程快速查询状态。目录必须预先存在、位于 GitNexus 包/构建根目录之外，且路径中不得包含符号链接或 junction。缺少 POSIX 所有权 API 的平台默认保持故障关闭。', 'help.analyze.environment': - '\n环境变量：\n GITNEXUS_NO_GITIGNORE=1 跳过 .gitignore 解析（仍读取 .gitnexusignore）\n GITNEXUS_MAX_FILE_SIZE=N 覆盖大文件跳过阈值（KB）。默认 512，最大 32768。\n GITNEXUS_ANALYZER_IDENTITY_CACHE_DIR=/absolute/protected/dir 由操作员明确信任的持久分析器身份缓存；目录必须预先存在、位于包/构建根目录之外，且路径中不得包含符号链接或 junction。\n GITNEXUS_WORKER_SUB_BATCH_TIMEOUT_MS=N Worker 空闲超时（毫秒）。默认 30000。\n GITNEXUS_WAL_CHECKPOINT_THRESHOLD=N LadybugDB WAL 自动 checkpoint 阈值（字节，默认 67108864 = 64 MiB；-1 保持 Ladybug 默认约 16 MiB）。\n GITNEXUS_WORKER_SUB_BATCH_MAX_BYTES=N Worker 作业字节预算。默认 8388608。\n GITNEXUS_WORKER_POOL_SIZE=N 解析 worker 数量覆盖值。默认 cores-1，最多 16。\n GITNEXUS_PARSE_CHUNK_CONCURRENCY=N 并发进行中的解析分块数。默认 2。\n GITNEXUS_WORKER_MAX_RESPAWNS_PER_SLOT=N 每个 slot 丢弃前允许的最大替换进程数。默认 3。\n GITNEXUS_WORKER_MAX_CUMULATIVE_TIMEOUT_MS=N 每个作业的总重试墙钟时间。默认 5 倍子批次超时。\n GITNEXUS_WORKER_CONSECUTIVE_FAILURE_THRESHOLD=N 每个 slot 触发熔断的死亡次数。默认 max(3, poolSize)。\n GITNEXUS_WORKER_SHUTDOWN_DRAIN_MS=N 线程池关闭时等待仍在原生代码中的已退役 worker 的最长时间（到达安全点后再终止，避免进程级 abort）。默认 30000。\n GITNEXUS_CPP_CAPTURE_BUDGET_MS=N C++ 捕获提取的每文件墙钟预算；超出后该文件保留部分捕获并输出警告。默认 20000。\n GITNEXUS_EMBEDDING_THREADS=N 限制 --embeddings 的本地 ONNX CPU 线程数。\n GITNEXUS_SEMANTIC_EXACT_SCAN_LIMIT=N exact-scan 回退的最大嵌入分块数。默认 10000。\n GITNEXUS_VECTOR_MAX_DISTANCE=N 语义/向量搜索接受的最大余弦距离（0 < N <= 2；超出则钳制为 2）。MCP 默认 0.6，其他路径默认 0.5。\n\n当参数和对应环境变量同时提供时，参数优先。\n\n提示：`.gitnexusignore` 支持 `.gitignore` 风格的取反。比如添加\n `!__tests__/` 可以索引默认自动过滤的目录（#771）。', + '\n环境变量：\n GITNEXUS_NO_GITIGNORE=1 跳过 .gitignore 解析（仍读取 .gitnexusignore）\n GITNEXUS_MAX_FILE_SIZE=N 覆盖大文件跳过阈值（KB）。默认 512，最大 32768。\n GITNEXUS_ANALYZER_IDENTITY_CACHE_DIR=/absolute/protected/dir 由操作员明确信任的持久分析器身份缓存；目录必须预先存在、位于包/构建根目录之外，且路径中不得包含符号链接或 junction。\n GITNEXUS_WORKER_SUB_BATCH_TIMEOUT_MS=N Worker 空闲超时（毫秒）。默认 30000。\n GITNEXUS_WAL_CHECKPOINT_THRESHOLD=N LadybugDB WAL 自动 checkpoint 阈值（字节，默认 67108864 = 64 MiB；-1 保持 Ladybug 默认约 16 MiB）。\n GITNEXUS_WORKER_SUB_BATCH_MAX_BYTES=N Worker 作业字节预算。默认 8388608。\n GITNEXUS_WORKER_POOL_SIZE=N 解析 worker 数量覆盖值。默认 cores-1，最多 16。\n GITNEXUS_PARSE_CHUNK_CONCURRENCY=N 并发进行中的解析分块数。默认 2。\n GITNEXUS_WORKER_MAX_RESPAWNS_PER_SLOT=N 每个 slot 丢弃前允许的最大替换进程数。默认 3。\n GITNEXUS_WORKER_MAX_CUMULATIVE_TIMEOUT_MS=N 每个作业的总重试墙钟时间。默认 5 倍子批次超时。\n GITNEXUS_WORKER_CONSECUTIVE_FAILURE_THRESHOLD=N 每个 slot 触发熔断的死亡次数。默认 max(3, poolSize)。\n GITNEXUS_WORKER_SHUTDOWN_DRAIN_MS=N 线程池关闭时等待仍在原生代码中的已退役 worker 的最长时间（到达安全点后再终止，避免进程级 abort）。默认 30000。\n GITNEXUS_CPP_CAPTURE_BUDGET_MS=N C++ 捕获提取的每文件墙钟预算；超出后该文件保留部分捕获并输出警告。默认 20000。\n GITNEXUS_EMBEDDING_THREADS=N 限制 --embeddings 的本地 ONNX CPU 线程数。\n GITNEXUS_EMBEDDING_RETRY_TIMEOUTS=1 将单次 HTTP 嵌入超时纳入 GITNEXUS_EMBEDDING_MAX_ATTEMPTS 重试（默认关闭，超时仍为终止错误）。\n GITNEXUS_SEMANTIC_EXACT_SCAN_LIMIT=N exact-scan 回退的最大嵌入分块数。默认 10000。\n GITNEXUS_VECTOR_MAX_DISTANCE=N 语义/向量搜索接受的最大余弦距离（0 < N <= 2；超出则钳制为 2）。MCP 默认 0.6，其他路径默认 0.5。\n\n当参数和对应环境变量同时提供时，参数优先。\n\n提示：`.gitnexusignore` 支持 `.gitignore` 风格的取反。比如添加\n `!__tests__/` 可以索引默认自动过滤的目录（#771）。', } satisfies EnglishMessages; diff --git a/gitnexus/src/cli/index.ts b/gitnexus/src/cli/index.ts index 70238dc00..fd1cb11c8 100644 --- a/gitnexus/src/cli/index.ts +++ b/gitnexus/src/cli/index.ts @@ -304,15 +304,13 @@ program .description('Install the latest published GitNexus globally (`npm i -g gitnexus@`).') .action(createLazyAction(() => import('./update.js'), 'updateCommand')); -program +const embeddings = program .command('embeddings') - .description('Manage the on-demand local embedding runtime') + .description(t('help.command.embeddings.description')); + +embeddings .command('install') - .description( - 'Install the local embedding stack (@huggingface/transformers + onnxruntime-node) on demand. ' + - 'Heals installs where npm skipped the optional packages (e.g. behind an HTTP proxy, #2370). ' + - 'Downloads only from your configured npm registry — mirrors and proxies apply.', - ) + .description(t('help.command.embeddings.install.description')) .option( '--cuda', "Also download the CUDA GPU binaries (runs onnxruntime-node's NuGet postinstall; " + @@ -321,6 +319,12 @@ program .option('--force', 'Install into the runtime prefix even when the stack already resolves') .action(createLazyAction(() => import('./embeddings.js'), 'embeddingsInstallCommand')); +embeddings + .command('sync [path]') + .description(t('help.command.embeddings.sync.description')) + .addHelpText('after', () => t('help.analyze.environment')) + .action(createLbugLazyAction(() => import('./embeddings-sync.js'), 'embeddingsSyncCommand')); + program .command('clean') .description('Delete GitNexus index for current repo') diff --git a/gitnexus/src/core/embeddings/http-client.ts b/gitnexus/src/core/embeddings/http-client.ts index 7919b2376..bc7458ea1 100644 --- a/gitnexus/src/core/embeddings/http-client.ts +++ b/gitnexus/src/core/embeddings/http-client.ts @@ -39,6 +39,7 @@ interface HttpConfig { retryCapMs: number; minIntervalMs: number; timeoutMs: number; + retryTimeouts: boolean; requestDimensions?: number; } @@ -202,6 +203,7 @@ const readConfig = (): HttpConfig | null => { DEFAULT_HTTP_TIMEOUT_MS, MAX_HTTP_TIMEOUT_MS, ), + retryTimeouts: parseNonNegativeIntegerEnv('GITNEXUS_EMBEDDING_RETRY_TIMEOUTS', 0, 1) === 1, requestDimensions, }; }; @@ -338,6 +340,36 @@ class RetryableEmbeddingBodyError extends Error { } } +class RetryableEmbeddingTimeoutError extends Error { + constructor( + readonly timeoutMs: number, + options?: { cause?: unknown }, + ) { + super( + `Embedding request timed out after ${timeoutMs}ms`, + options?.cause !== undefined ? { cause: options.cause } : undefined, + ); + this.name = 'RetryableEmbeddingTimeoutError'; + } +} + +/** Re-wrap an opt-in TimeoutError so `resilientFetch` retries it. Abort stays terminal. */ +const throwIfRetryableTimeout = ( + err: unknown, + retryTimeouts: boolean, + callerAborted: boolean | undefined, + timeoutMs: number, +): void => { + if ( + retryTimeouts && + !callerAborted && + isTerminalNetworkError(err) && + err.name === 'TimeoutError' + ) { + throw new RetryableEmbeddingTimeoutError(timeoutMs, { cause: err }); + } +}; + /** * Build the message for a 2xx body carrying the wrong number of vectors. * @@ -384,6 +416,7 @@ const httpEmbedBatch = async ( retryCapMs = HTTP_RETRY_CAP_MS, minIntervalMs = 0, timeoutMs = DEFAULT_HTTP_TIMEOUT_MS, + retryTimeouts = false, ): Promise => { const requestBody: { input: string[]; model: string; dimensions?: number } = { input: batch, @@ -428,7 +461,13 @@ const httpEmbedBatch = async ( const signal = requestOptions.signal ? AbortSignal.any([requestOptions.signal, timeoutSignal]) : timeoutSignal; - const attemptResp = await globalThis.fetch(input, { ...init, signal }); + let attemptResp: Response; + try { + attemptResp = await globalThis.fetch(input, { ...init, signal }); + } catch (err) { + throwIfRetryableTimeout(err, retryTimeouts, requestOptions.signal?.aborted, timeoutMs); + throw err; + } // Non-OK bodies are none of our business: hand the response straight // back so `resilientFetch` keeps classifying 4xx/5xx/429 unchanged. if (!attemptResp.ok) return attemptResp; @@ -447,15 +486,19 @@ const httpEmbedBatch = async ( // Not every `.json()` rejection is a parse error: the per-attempt // signal (`AbortSignal.any([caller, AbortSignal.timeout(...)])`) is // wired to the body stream, so a stalled body rejects with the abort - // reason. Re-raise those untouched — `isTerminalNetworkError` is - // `resilientFetch`'s own predicate, so this test agrees with - // `classifyOutcome` by construction. Wrapping one would flip its - // verdict from `terminal-network` (returned without retry AND - // without touching the breaker, via `recordNeutral()`) to - // `retryable-network` (retried, then `breaker.recordFailure()`): the - // same timeout would take 3 attempts instead of 1, count toward the - // process-global `embeddings-http` breaker, and reach the operator as - // "unparseable response" so they never reach for the timeout knob. + // reason. Re-raise AbortError (and TimeoutError when retry is off) + // untouched — `isTerminalNetworkError` is `resilientFetch`'s own + // predicate, so this test agrees with `classifyOutcome` by + // construction. Wrapping one would flip its verdict from + // `terminal-network` (returned without retry AND without touching + // the breaker, via `recordNeutral()`) to `retryable-network` + // (retried, then `breaker.recordFailure()`): the same timeout would + // take 3 attempts instead of 1, count toward the process-global + // `embeddings-http` breaker, and reach the operator as "unparseable + // response" so they never reach for the timeout knob. + // Opt-in `GITNEXUS_EMBEDDING_RETRY_TIMEOUTS=1` is the exception: + // TimeoutError is re-wrapped so the existing retry loop can retry it. + throwIfRetryableTimeout(err, retryTimeouts, requestOptions.signal?.aborted, timeoutMs); if (isTerminalNetworkError(err)) throw err; throw new RetryableEmbeddingBodyError(unparseableMessage(), { cause: err }); } @@ -503,6 +546,12 @@ const httpEmbedBatch = async ( if (err instanceof RetryableEmbeddingBodyError) { throw new HttpEmbeddingError(err.terminalMessage, { cause: err.cause }); } + if (err instanceof RetryableEmbeddingTimeoutError) { + throw new HttpEmbeddingError( + `${err.message} after ${maxAttempts} attempt(s) (${safeUrl(url)}, batch ${batchIndex})`, + { cause: err.cause }, + ); + } if (err instanceof CircuitOpenError) { throw new HttpEmbeddingError( `Embedding endpoint circuit open (${safeUrl(url)}, batch ${batchIndex}): retry in ${Math.ceil(err.retryAfterMs / 1000)}s`, @@ -580,6 +629,7 @@ export const httpEmbed = async ( config.retryCapMs, config.minIntervalMs, config.timeoutMs, + config.retryTimeouts, ); // Defensive backstop, deliberately kept: `httpEmbedBatch` now rejects a @@ -655,6 +705,7 @@ export const httpEmbedQuery = async ( config.retryCapMs, config.minIntervalMs, config.timeoutMs, + config.retryTimeouts, ); // Defensive backstop like the `httpEmbed` one above: an empty `data` array is // now a cardinality mismatch (0 vectors for 1 text) rejected and retried diff --git a/gitnexus/test/unit/cli-index-help.test.ts b/gitnexus/test/unit/cli-index-help.test.ts index 4911af27f..51f187ab9 100644 --- a/gitnexus/test/unit/cli-index-help.test.ts +++ b/gitnexus/test/unit/cli-index-help.test.ts @@ -56,6 +56,7 @@ const allHelpCommands = [ ['eval-server'], ['embeddings'], ['embeddings', 'install'], + ['embeddings', 'sync'], ['group'], ['group', 'create'], ['group', 'add'], diff --git a/gitnexus/test/unit/embeddings-sync-command.test.ts b/gitnexus/test/unit/embeddings-sync-command.test.ts new file mode 100644 index 000000000..71e75f558 --- /dev/null +++ b/gitnexus/test/unit/embeddings-sync-command.test.ts @@ -0,0 +1,272 @@ +/** + * Tests for `gitnexus embeddings sync` writer-safety contracts (#3065 review): + * index lock, missing-DB preflight, identity fail-closed, tri-state count, + * closeLbug masking, and hash-only cache load. + */ +import { mkdtemp, mkdir, rm, writeFile } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import path from 'node:path'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; + +const { + acquireIndexLockMock, + releaseMock, + getStoragePathsMock, + loadMetaMock, + saveMetaMock, + initLbugMock, + closeLbugMock, + executeQueryMock, + executeWithReusedStatementMock, + fetchExistingEmbeddingHashesMock, + runEmbeddingPipelineMock, + resolveEmbeddingIdentityMock, +} = vi.hoisted(() => ({ + acquireIndexLockMock: vi.fn(), + releaseMock: vi.fn(), + getStoragePathsMock: vi.fn(), + loadMetaMock: vi.fn(), + saveMetaMock: vi.fn(), + initLbugMock: vi.fn(), + closeLbugMock: vi.fn(), + executeQueryMock: vi.fn(), + executeWithReusedStatementMock: vi.fn(), + fetchExistingEmbeddingHashesMock: vi.fn(), + runEmbeddingPipelineMock: vi.fn(), + resolveEmbeddingIdentityMock: vi.fn(), +})); + +vi.mock('../../src/storage/git.js', () => ({ + getGitRoot: () => '/tmp/emb-sync-repo', +})); + +vi.mock('../../src/storage/index-lock.js', () => ({ + acquireIndexLock: (...args: unknown[]) => acquireIndexLockMock(...args), +})); + +vi.mock('../../src/storage/repo-manager.js', () => ({ + getStoragePaths: (...args: unknown[]) => getStoragePathsMock(...args), + loadMeta: (...args: unknown[]) => loadMetaMock(...args), + saveMeta: (...args: unknown[]) => saveMetaMock(...args), +})); + +vi.mock('../../src/core/lbug/lbug-adapter.js', () => ({ + initLbug: (...args: unknown[]) => initLbugMock(...args), + closeLbug: (...args: unknown[]) => closeLbugMock(...args), + executeQuery: (...args: unknown[]) => executeQueryMock(...args), + executeWithReusedStatement: (...args: unknown[]) => executeWithReusedStatementMock(...args), + fetchExistingEmbeddingHashes: (...args: unknown[]) => fetchExistingEmbeddingHashesMock(...args), +})); + +vi.mock('../../src/core/embeddings/embedding-pipeline.js', () => ({ + runEmbeddingPipeline: (...args: unknown[]) => runEmbeddingPipelineMock(...args), +})); + +vi.mock('../../src/core/embeddings/embedding-identity.js', () => ({ + resolveEmbeddingIdentity: () => resolveEmbeddingIdentityMock(), +})); + +const IDENTITY = { model: 'test-model', dimensions: 768, provider: 'local' } as const; + +const BASE_META = { + repoPath: '/tmp/emb-sync-repo', + lastCommit: 'abc123', + indexedAt: '2026-01-01T00:00:00.000Z', + stats: { embeddings: 1 }, +}; + +const lockHandle = (release: () => void = releaseMock) => ({ + record: { + v: 1 as const, + pid: 1, + hostname: 'h', + startTime: null, + token: 't', + invocationId: 'i', + acquiredAt: '', + }, + release, +}); + +async function run(inputPath = '/tmp/emb-sync-repo') { + const { embeddingsSyncCommand } = await import('../../src/cli/embeddings-sync.js'); + await embeddingsSyncCommand(inputPath); +} + +describe('embeddingsSyncCommand writer safety (#3065)', () => { + const tmpDirs: string[] = []; + + async function store(kind: 'file' | 'missing' | 'dir' = 'file') { + const dir = await mkdtemp(path.join(tmpdir(), 'emb-sync-')); + tmpDirs.push(dir); + const lbugPath = path.join(dir, 'lbug'); + const metaPath = path.join(dir, 'gitnexus.json'); + if (kind === 'file') await writeFile(lbugPath, 'db'); + if (kind === 'dir') await mkdir(lbugPath); + getStoragePathsMock.mockReturnValue({ lbugPath, metaPath }); + return { dir, lbugPath, metaPath }; + } + + beforeEach(() => { + vi.resetModules(); + acquireIndexLockMock.mockReset().mockResolvedValue(lockHandle()); + releaseMock.mockReset(); + getStoragePathsMock.mockReset(); + loadMetaMock.mockReset().mockResolvedValue({ ...BASE_META }); + saveMetaMock.mockReset().mockResolvedValue(undefined); + initLbugMock.mockReset().mockResolvedValue(undefined); + closeLbugMock.mockReset().mockResolvedValue(undefined); + executeQueryMock.mockReset().mockResolvedValue([{ cnt: 2 }]); + executeWithReusedStatementMock.mockReset(); + fetchExistingEmbeddingHashesMock.mockReset().mockResolvedValue(new Map([['n1', 'hash-1']])); + runEmbeddingPipelineMock.mockReset().mockResolvedValue({ + nodesProcessed: 2, + chunksProcessed: 2, + failedNodeIds: [], + }); + resolveEmbeddingIdentityMock.mockReset().mockReturnValue({ ...IDENTITY }); + }); + + afterEach(async () => { + await Promise.all(tmpDirs.splice(0).map((dir) => rm(dir, { recursive: true, force: true }))); + }); + + it('acquires the index lock before re-reading metadata and releases it in finally', async () => { + const { dir } = await store(); + const order: string[] = []; + acquireIndexLockMock.mockImplementation(async () => { + order.push('lock'); + return lockHandle(() => { + order.push('release'); + releaseMock(); + }); + }); + loadMetaMock.mockImplementation(async () => { + order.push('loadMeta'); + return { ...BASE_META }; + }); + initLbugMock.mockImplementation(async () => { + order.push('init'); + }); + + await run(); + + expect(acquireIndexLockMock).toHaveBeenCalledWith(dir); + expect(order[0]).toBe('lock'); + expect(order.indexOf('loadMeta')).toBeGreaterThan(order.indexOf('lock')); + expect(order.indexOf('init')).toBeGreaterThan(order.indexOf('loadMeta')); + expect(order.at(-1)).toBe('release'); + }); + + it('refuses to create a new database when the LadybugDB file is missing', async () => { + const { lbugPath } = await store('missing'); + await expect(run()).rejects.toThrow( + `The LadybugDB graph store at ${lbugPath} is missing. Run gitnexus analyze first.`, + ); + expect(initLbugMock).not.toHaveBeenCalled(); + expect(releaseMock).toHaveBeenCalled(); + }); + + it('refuses to open a LadybugDB path that is not a regular file', async () => { + const { lbugPath } = await store('dir'); + await expect(run()).rejects.toThrow( + `The LadybugDB graph store at ${lbugPath} is not a usable database file. Run gitnexus analyze first.`, + ); + expect(initLbugMock).not.toHaveBeenCalled(); + expect(releaseMock).toHaveBeenCalled(); + }); + + it('fails closed on an identity-mismatched partial checkpoint', async () => { + await store(); + loadMetaMock.mockResolvedValue({ + ...BASE_META, + embeddingCheckpoint: { + at: '2026-01-01T00:00:00.000Z', + nodesProcessed: 1, + totalNodes: 2, + chunksProcessed: 1, + model: 'old-model', + dimensions: 768, + provider: 'local', + kind: 'partial', + pendingNodeIds: ['n2'], + }, + }); + resolveEmbeddingIdentityMock.mockReturnValue({ + model: 'new-model', + dimensions: 768, + provider: 'http:deadbeef', + }); + + await expect(run()).rejects.toThrow(/Cannot sync embeddings: the index checkpoint was written/); + expect(initLbugMock).not.toHaveBeenCalled(); + expect(releaseMock).toHaveBeenCalled(); + }); + + it('allows an unverified-count checkpoint under a different identity', async () => { + await store(); + loadMetaMock.mockResolvedValue({ + ...BASE_META, + embeddingCheckpoint: { + at: '2026-01-01T00:00:00.000Z', + nodesProcessed: 2, + totalNodes: 2, + chunksProcessed: 2, + model: 'old-model', + dimensions: 768, + provider: 'local', + kind: 'unverified-count', + pendingNodeIds: [], + }, + }); + resolveEmbeddingIdentityMock.mockReturnValue({ + model: 'new-model', + dimensions: 768, + provider: 'http:deadbeef', + }); + + await run(); + expect(initLbugMock).toHaveBeenCalled(); + expect(runEmbeddingPipelineMock).toHaveBeenCalled(); + }); + + it('does not publish a missing count cell as zero', async () => { + await store(); + executeQueryMock.mockResolvedValue([{}]); + + await expect(run()).rejects.toThrow('Could not verify persisted embedding count.'); + expect(saveMetaMock).toHaveBeenCalledTimes(1); + const saved = saveMetaMock.mock.calls[0]?.[1] as { + stats?: { embeddings?: number }; + embeddingCheckpoint?: { kind?: string; pendingNodeIds?: string[] }; + }; + expect(saved.stats?.embeddings).toBe(1); + expect(saved.embeddingCheckpoint?.kind).toBe('unverified-count'); + expect(saved.embeddingCheckpoint?.pendingNodeIds).toEqual([]); + expect(releaseMock).toHaveBeenCalled(); + }); + + it('keeps the pipeline error when closeLbug also rejects', async () => { + await store(); + runEmbeddingPipelineMock.mockRejectedValue(new Error('pipeline boom')); + closeLbugMock.mockRejectedValue(new Error('close boom')); + + await expect(run()).rejects.toThrow('pipeline boom'); + expect(releaseMock).toHaveBeenCalled(); + }); + + it('loads existing hashes without materializing cached vectors', async () => { + await store(); + const hashes = new Map([ + ['n1', 'h1'], + ['n2', 'h2'], + ]); + fetchExistingEmbeddingHashesMock.mockResolvedValue(hashes); + + await run(); + + expect(fetchExistingEmbeddingHashesMock).toHaveBeenCalledTimes(1); + const existingArg = runEmbeddingPipelineMock.mock.calls[0]?.[5]; + expect(existingArg).toBe(hashes); + }); +}); diff --git a/gitnexus/test/unit/http-embedder.test.ts b/gitnexus/test/unit/http-embedder.test.ts index 397481569..5daf565f6 100644 --- a/gitnexus/test/unit/http-embedder.test.ts +++ b/gitnexus/test/unit/http-embedder.test.ts @@ -11,6 +11,7 @@ const ENV_KEYS = [ 'GITNEXUS_EMBEDDING_RETRY_CAP_MS', 'GITNEXUS_EMBEDDING_MIN_INTERVAL_MS', 'GITNEXUS_EMBEDDING_REQUEST_DIMS', + 'GITNEXUS_EMBEDDING_RETRY_TIMEOUTS', ] as const; /** 384d mock vector matching the default schema dimensions. */ @@ -733,6 +734,27 @@ describe('HTTP embedding backend', () => { expect(fetch).toHaveBeenCalledTimes(1); }); + it('retries a timeout when explicitly configured', async () => { + process.env.GITNEXUS_EMBEDDING_URL = 'http://test:8080/v1'; + process.env.GITNEXUS_EMBEDDING_MODEL = 'test-model'; + process.env.GITNEXUS_EMBEDDING_RETRY_TIMEOUTS = '1'; + process.env.GITNEXUS_EMBEDDING_MAX_ATTEMPTS = '2'; + process.env.GITNEXUS_EMBEDDING_RETRY_CAP_MS = '1'; + + const timeoutErr = new DOMException( + 'The operation was aborted due to timeout', + 'TimeoutError', + ); + const ok = { ok: true, json: async () => ({ data: [{ embedding: mockVec }] }) }; + vi.stubGlobal('fetch', vi.fn().mockRejectedValueOnce(timeoutErr).mockResolvedValueOnce(ok)); + + const { embedText } = await import('../../src/core/embeddings/embedder.js'); + const result = await embedText('test'); + + expect(fetch).toHaveBeenCalledTimes(2); + expect(result).toBeInstanceOf(Float32Array); + }); + it('retries on network error then succeeds', async () => { process.env.GITNEXUS_EMBEDDING_URL = 'http://test:8080/v1'; process.env.GITNEXUS_EMBEDDING_MODEL = 'test-model';