feat(analyze): add incremental watch mode (#3072)

* feat(analyze): add incremental watch mode

* fix(watch): harden control file reads

* fix(watch): contain refresh errors and bound reads

* fix(watch): stream strict control file reads

* fix(watch): harden refresh recovery and lifecycle

* fix(watch): report ignored repository defaults

* fix(analyze): preserve signal exit semantics

* style(analyze): format signal exit helper

* test(config): exercise descriptor growth guard

* test(watch): await source event before rename

* fix(watch): keep live-index retries honest and ignore analyzer writes

Hold retry backoff when events merge, stop only after a live-index mutation, skip .gitnexus self-writes, and reject the remaining one-shot watch flags. Export impact-risk scoring from gitnexus-shared so consumers can share the same scale.

Co-authored-by: Cursor <cursoragent@cursor.com>

* fix(watch): contain queue edge cases after review

Preserve overflow-only refreshes, contain synchronous refresh failures, and mark successful atomic publication before later operations can fail.

Co-authored-by: Cursor <cursoragent@cursor.com>

---------

Co-authored-by: Gergo Magyar <gergomagyar0@gmail.com>
Co-authored-by: Cursor <cursoragent@cursor.com>
Co-authored-by: Gergő Magyar <gergomagyar@icloud.com>
This commit is contained in:
azizur100389 2026-08-29 11:40:56 +01:00 committed by GitHub
parent 7c723ce794
commit bf7dcf98ca
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
39 changed files with 2504 additions and 89 deletions

View file

@ -21,6 +21,8 @@ Run from the project root. This parses all source files, builds the knowledge gr
| Flag | Effect |
| -------------- | ---------------------------------------------------------------- |
| `--watch` | Keep a Git repository index current with serialized refreshes |
| `--debounce <ms>` | Watch quiet period before refresh (default: 300 ms) |
| `--force` | Force full re-index even if up to date |
| `--embeddings` | Enable embedding generation for semantic search (off by default) |
| `--drop-embeddings` | Drop existing embeddings on rebuild. By default, an `analyze` without `--embeddings` preserves them. |
@ -28,6 +30,8 @@ Run from the project root. This parses all source files, builds the knowledge gr
**When to run:** First time in a project, after major code changes, or when `gitnexus://repo/{name}/context` reports the index is stale. In Claude Code, a PostToolUse hook detects staleness after `git commit` and `git merge` and notifies the agent to run `analyze` — the hook does not run analyze itself, to avoid blocking the agent for up to 120s and risking KuzuDB corruption on timeout.
Use `node .gitnexus/run.cjs analyze --watch` for a long-lived local Git repository. It performs an initial analysis, queues scanner-admitted file changes, and retries intact failed batches with bounded backoff. Watch refreshes update only the graph: they skip AGENTS.md / CLAUDE.md injection and standard skill installation, so run a one-shot `analyze` when those generated files need updating. Watch rejects one-shot or context-output flags including `--force`, embedding flags, `--skills`, `--default-branch`, `--skip-agents-md`, `--skip-skills`, `--no-stats`, `--self-commit`, `--index-only`, and `--skip-git`. It never pulls remotes. Running MCP and `serve` processes periodically check for a published replacement and reopen it without a restart. MCP checks are throttled to once every five seconds, so a tool call before the next check can briefly use the previous index.
### status — Check index freshness
```bash
@ -86,5 +90,5 @@ Lists all repositories registered in `~/.gitnexus/registry.json`. The MCP `list_
## Troubleshooting
- **"Not inside a git repository"**: Run from a directory inside a git repo
- **Index is stale after re-analyzing**: Restart Claude Code to reload the MCP server
- **Index is stale after re-analyzing**: Wait for the next MCP tool call to reopen the published index; this normally takes no more than five seconds
- **Embeddings slow**: Omit `--embeddings` (it's off by default) or set `OPENAI_API_KEY` for faster API-based embedding

View file

@ -384,6 +384,7 @@ Everyday commands:
```bash
gitnexus setup # Configure MCP for detected editors (one-time; -c to select)
gitnexus analyze [path] # Index a repository (or update a stale index)
gitnexus analyze [path] --watch # Watch local files and serialize incremental refreshes
gitnexus mcp # Start MCP server (stdio) — serves all indexed repos
gitnexus serve # Start local HTTP server (multi-repo) for web UI connection
gitnexus eval-server # Start lightweight evaluation HTTP tools (loopback by default)
@ -396,6 +397,28 @@ gitnexus uninstall # Preview removal of GitNexus MCP/skills/hooks
You can also query the graph directly from the terminal — `gitnexus query`, `context`, `impact`, `trace`, `cypher`, `detect-changes`, and `check` mirror the MCP tools of the same names, and `gitnexus doctor` prints runtime platform capabilities.
`gitnexus analyze --watch` requires a Git repository. It runs one initial
analysis, then debounces scanner-admitted working-tree changes for 300 ms by
default and applies serialized incremental refreshes. Events arriving during a
refresh remain queued, and retryable failures retain the same batch with bounded
backoff. Invalid `.gitnexusrc` or ignore-file reloads pause ordinary refreshes
until the control file is fixed. Stop the watcher with Ctrl+C.
Watch mode accepts `--debounce`, `--workers`, `--worker-timeout`,
`--max-file-size`, `--branch`, `--pdg`, `--name`, `--allow-duplicate-name`, and
`--verbose`. Explicit one-shot options such as `--force`, `--repair-fts`,
embedding flags, `--skills`, `--self-commit`, `--index-only`, and `--skip-git`
are rejected. Unsupported defaults from `.gitnexusrc` are ignored with a
warning rather than making an otherwise valid repository unwatchable.
POSIX requests clone-first copy-and-swap publication when the live index has no
orphan sidecars. Windows and sidecar fallback runs update in place: failures
known to occur before writes are retried, while a failure that may have mutated
the live index stops the watcher. Watch mode does not pull remotes. Running MCP
and `serve` processes reopen a newly published index automatically; MCP observes
the replacement on its next tool call, typically within five seconds, so no
restart is required.
<details>
<summary><strong>Authenticated <code>eval-server</code> binding</strong></summary>

View file

@ -21,6 +21,8 @@ Run from the project root. This parses all source files, builds the knowledge gr
| Flag | Effect |
| -------------- | ---------------------------------------------------------------- |
| `--watch` | Keep a Git repository index current with serialized refreshes |
| `--debounce <ms>` | Watch quiet period before refresh (default: 300 ms) |
| `--force` | Force full re-index even if up to date |
| `--embeddings` | Enable embedding generation for semantic search (off by default) |
| `--drop-embeddings` | Drop existing embeddings on rebuild. By default, an `analyze` without `--embeddings` preserves them. |
@ -28,6 +30,8 @@ Run from the project root. This parses all source files, builds the knowledge gr
**When to run:** First time in a project, after major code changes, or when `gitnexus://repo/{name}/context` reports the index is stale. In Claude Code, a PostToolUse hook detects staleness after `git commit` and `git merge` and notifies the agent to run `analyze` — the hook does not run analyze itself, to avoid blocking the agent for up to 120s and risking KuzuDB corruption on timeout.
Use `node .gitnexus/run.cjs analyze --watch` for a long-lived local Git repository. It performs an initial analysis, queues scanner-admitted file changes, and retries intact failed batches with bounded backoff. Watch refreshes update only the graph: they skip AGENTS.md / CLAUDE.md injection and standard skill installation, so run a one-shot `analyze` when those generated files need updating. Watch rejects one-shot or context-output flags including `--force`, embedding flags, `--skills`, `--default-branch`, `--skip-agents-md`, `--skip-skills`, `--no-stats`, `--self-commit`, `--index-only`, and `--skip-git`. It never pulls remotes. Running MCP and `serve` processes periodically check for a published replacement and reopen it without a restart. MCP checks are throttled to once every five seconds, so a tool call before the next check can briefly use the previous index.
### status — Check index freshness
```bash
@ -86,5 +90,5 @@ Lists all repositories registered in `~/.gitnexus/registry.json`. The MCP `list_
## Troubleshooting
- **"Not inside a git repository"**: Run from a directory inside a git repo
- **Index is stale after re-analyzing**: Restart Claude Code to reload the MCP server
- **Index is stale after re-analyzing**: Wait for the next MCP tool call to reopen the published index; this normally takes no more than five seconds
- **Embeddings slow**: Omit `--embeddings` (it's off by default) or set `OPENAI_API_KEY` for faster API-based embedding

View file

@ -0,0 +1,129 @@
export type ImpactRisk = 'LOW' | 'MEDIUM' | 'HIGH' | 'CRITICAL' | 'UNKNOWN';
export type ImpactRiskAxis = 'processes' | 'modules';
export type UnusedImpactRiskReason =
| 'file-nodes-have-no-process-or-community-membership'
| 'enrichment-skipped'
| 'enrichment-budget-exhausted'
| 'enrichment-query-failed';
export interface UnusedImpactRiskAxis {
axis: ImpactRiskAxis;
reason: UnusedImpactRiskReason;
}
export interface ImpactRiskInput {
direction: 'upstream' | 'downstream';
directCount: number;
processCount: number;
moduleCount: number;
impactedCount: number;
unusedAxes?: readonly UnusedImpactRiskAxis[];
}
export interface ImpactRiskResult {
risk: ImpactRisk;
riskSharedAxes: ImpactRisk;
riskScale: {
comparableAcrossKinds: boolean;
unusedAxes: readonly UnusedImpactRiskAxis[];
};
}
function score(
input: Pick<
ImpactRiskInput,
'direction' | 'directCount' | 'processCount' | 'moduleCount' | 'impactedCount'
>,
): ImpactRisk {
const { direction, directCount, processCount, moduleCount, impactedCount } = input;
if (direction === 'upstream' && impactedCount === 0) return 'UNKNOWN';
if (directCount >= 30 || processCount >= 5 || moduleCount >= 5 || impactedCount >= 200) {
return 'CRITICAL';
}
if (directCount >= 15 || processCount >= 3 || moduleCount >= 3 || impactedCount >= 100) {
return 'HIGH';
}
if (directCount >= 5 || impactedCount >= 30) return 'MEDIUM';
return 'LOW';
}
function countsWithUnusedAxesZeroed(
input: ImpactRiskInput,
): Pick<
ImpactRiskInput,
'direction' | 'directCount' | 'processCount' | 'moduleCount' | 'impactedCount'
> {
let processCount = input.processCount;
let moduleCount = input.moduleCount;
for (const unused of input.unusedAxes ?? []) {
if (unused.axis === 'processes') processCount = 0;
if (unused.axis === 'modules') moduleCount = 0;
}
return {
direction: input.direction,
directCount: input.directCount,
processCount,
moduleCount,
impactedCount: input.impactedCount,
};
}
/** Map walk outcomes to unused process/module axes so comparability matches what was sampled. */
export function unusedAxesForImpactWalk(input: {
isFileTarget: boolean;
skipEnrichment: boolean;
maxChunks: number;
processQueryFailed: boolean;
moduleQueryFailed: boolean;
/** When 0, a zero chunk budget is not an unused-axis event — there was nothing to enrich. */
impactedCount?: number;
}): UnusedImpactRiskAxis[] {
if (input.isFileTarget) {
return [
{
axis: 'processes',
reason: 'file-nodes-have-no-process-or-community-membership',
},
{
axis: 'modules',
reason: 'file-nodes-have-no-process-or-community-membership',
},
];
}
if (input.skipEnrichment) {
return [
{ axis: 'processes', reason: 'enrichment-skipped' },
{ axis: 'modules', reason: 'enrichment-skipped' },
];
}
if (input.maxChunks === 0 && (input.impactedCount ?? 1) > 0) {
return [
{ axis: 'processes', reason: 'enrichment-budget-exhausted' },
{ axis: 'modules', reason: 'enrichment-budget-exhausted' },
];
}
const unused: UnusedImpactRiskAxis[] = [];
if (input.processQueryFailed) {
unused.push({ axis: 'processes', reason: 'enrichment-query-failed' });
}
if (input.moduleQueryFailed) {
unused.push({ axis: 'modules', reason: 'enrichment-query-failed' });
}
return unused;
}
export function scoreImpactRisk(input: ImpactRiskInput): ImpactRiskResult {
const unusedAxes = input.unusedAxes ?? [];
return {
risk: score(countsWithUnusedAxesZeroed(input)),
riskSharedAxes: score({ ...input, processCount: 0, moduleCount: 0 }),
riskScale: {
comparableAcrossKinds: unusedAxes.length === 0,
unusedAxes,
},
};
}

View file

@ -25,6 +25,17 @@ export {
} from './language-detection.js';
export type { MroStrategy } from './mro-strategy.js';
// Impact risk scoring
export { scoreImpactRisk, unusedAxesForImpactWalk } from './impact-risk.js';
export type {
ImpactRisk,
ImpactRiskAxis,
ImpactRiskInput,
ImpactRiskResult,
UnusedImpactRiskAxis,
UnusedImpactRiskReason,
} from './impact-risk.js';
// Pipeline progress
export type { PipelinePhase, PipelineProgress } from './pipeline.js';

View file

@ -234,6 +234,7 @@ Your AI agent gets **17 tools** (15 per-repo + 2 group) automatically:
gitnexus setup # Configure MCP for detected editors (one-time; use -c to select)
gitnexus uninstall # Preview removal of GitNexus MCP/skills/hooks (add --force to apply)
gitnexus analyze [path] # Index a repository (or update stale index)
gitnexus analyze [path] --watch # Watch local files and serialize incremental refreshes
gitnexus analyze --repair-fts # Fast path: rebuild/verify only FTS indexes on existing index data
gitnexus analyze --force # Full rebuild: re-parse + graph rebuild + FTS rebuild
gitnexus analyze --embeddings # Enable embedding generation (slower, better search)
@ -282,6 +283,31 @@ gitnexus group status <name> # Check staleness of repos in a group
gitnexus group impact <name> --target <symbol> --repo <groupPath> # Cross-repo blast radius
```
`gitnexus analyze --watch` requires a Git repository. It performs an initial
analysis and then debounces scanner-admitted working-tree changes for 300 ms by
default into serialized incremental refreshes. Events arriving during a run
remain queued, and retryable failures retain the same batch with bounded
backoff. Invalid `.gitnexusrc` or ignore-file reloads pause ordinary refreshes
until the control file is fixed. Watch refreshes update only the graph: they
intentionally skip AGENTS.md / CLAUDE.md injection and standard skill
installation. Run a one-shot `gitnexus analyze` when those generated files need
updating. Stop watch mode with Ctrl+C.
Watch mode accepts `--debounce`, `--workers`, `--worker-timeout`,
`--max-file-size`, `--branch`, `--pdg`, `--name`, `--allow-duplicate-name`, and
`--verbose`. Explicit one-shot options such as `--force`, `--repair-fts`,
embedding flags, `--skills`, `--default-branch`, `--skip-agents-md`,
`--skip-skills`, `--no-stats`, `--self-commit`, `--index-only`, and `--skip-git`
are rejected. Unsupported defaults from `.gitnexusrc` are ignored with a warning.
POSIX requests clone-first copy-and-swap publication when the live index has no
orphan sidecars. Windows and sidecar fallback runs update in place: failures
known to occur before writes are retried, while a failure that may have mutated
the live index stops the watcher. Watch mode does not pull remotes. Running MCP
and `serve` processes periodically check for a newly published index and reopen
it without a restart. MCP checks are throttled to once every five seconds, so a
tool call before the next check can briefly use the previous index.
GraphQL contract matching is opt-in in the group's `group.yaml`:
```yaml

View file

@ -14,6 +14,7 @@
"@modelcontextprotocol/sdk": "^1.0.0",
"@scarf/scarf": "^1.4.0",
"busboy": "^1.6.0",
"chokidar": "^4.0.3",
"cli-progress": "^3.12.0",
"commander": "^15.0.0",
"cors": "^2.8.5",
@ -826,9 +827,6 @@
"cpu": [
"arm"
],
"libc": [
"glibc"
],
"license": "LGPL-3.0-or-later",
"optional": true,
"os": [
@ -845,9 +843,6 @@
"cpu": [
"arm64"
],
"libc": [
"glibc"
],
"license": "LGPL-3.0-or-later",
"optional": true,
"os": [
@ -864,9 +859,6 @@
"cpu": [
"ppc64"
],
"libc": [
"glibc"
],
"license": "LGPL-3.0-or-later",
"optional": true,
"os": [
@ -883,9 +875,6 @@
"cpu": [
"riscv64"
],
"libc": [
"glibc"
],
"license": "LGPL-3.0-or-later",
"optional": true,
"os": [
@ -902,9 +891,6 @@
"cpu": [
"s390x"
],
"libc": [
"glibc"
],
"license": "LGPL-3.0-or-later",
"optional": true,
"os": [
@ -921,9 +907,6 @@
"cpu": [
"x64"
],
"libc": [
"glibc"
],
"license": "LGPL-3.0-or-later",
"optional": true,
"os": [
@ -940,9 +923,6 @@
"cpu": [
"arm64"
],
"libc": [
"musl"
],
"license": "LGPL-3.0-or-later",
"optional": true,
"os": [
@ -959,9 +939,6 @@
"cpu": [
"x64"
],
"libc": [
"musl"
],
"license": "LGPL-3.0-or-later",
"optional": true,
"os": [
@ -978,9 +955,6 @@
"cpu": [
"arm"
],
"libc": [
"glibc"
],
"license": "Apache-2.0",
"optional": true,
"os": [
@ -1003,9 +977,6 @@
"cpu": [
"arm64"
],
"libc": [
"glibc"
],
"license": "Apache-2.0",
"optional": true,
"os": [
@ -1028,9 +999,6 @@
"cpu": [
"ppc64"
],
"libc": [
"glibc"
],
"license": "Apache-2.0",
"optional": true,
"os": [
@ -1053,9 +1021,6 @@
"cpu": [
"riscv64"
],
"libc": [
"glibc"
],
"license": "Apache-2.0",
"optional": true,
"os": [
@ -1078,9 +1043,6 @@
"cpu": [
"s390x"
],
"libc": [
"glibc"
],
"license": "Apache-2.0",
"optional": true,
"os": [
@ -1103,9 +1065,6 @@
"cpu": [
"x64"
],
"libc": [
"glibc"
],
"license": "Apache-2.0",
"optional": true,
"os": [
@ -1128,9 +1087,6 @@
"cpu": [
"arm64"
],
"libc": [
"musl"
],
"license": "Apache-2.0",
"optional": true,
"os": [
@ -1153,9 +1109,6 @@
"cpu": [
"x64"
],
"libc": [
"musl"
],
"license": "Apache-2.0",
"optional": true,
"os": [
@ -2429,6 +2382,21 @@
"url": "https://github.com/chalk/chalk-template?sponsor=1"
}
},
"node_modules/chokidar": {
"version": "4.0.3",
"resolved": "https://registry.npmjs.org/chokidar/-/chokidar-4.0.3.tgz",
"integrity": "sha512-Qgzu8kfBvo+cA4962jnP1KkS6Dop5NS6g7R5LFYJr4b8Ub94PPQXUksCw9PvXoeXPRRddRNC5C1JQUR2SMGtnA==",
"license": "MIT",
"dependencies": {
"readdirp": "^4.0.1"
},
"engines": {
"node": ">= 14.16.0"
},
"funding": {
"url": "https://paulmillr.com/funding/"
}
},
"node_modules/chownr": {
"version": "3.0.0",
"resolved": "https://registry.npmjs.org/chownr/-/chownr-3.0.0.tgz",
@ -4659,6 +4627,19 @@
"rc": "cli.js"
}
},
"node_modules/readdirp": {
"version": "4.1.2",
"resolved": "https://registry.npmjs.org/readdirp/-/readdirp-4.1.2.tgz",
"integrity": "sha512-GDhwkLfywWL2s6vEjyhri+eXmfH6j1L7JE27WhqLeYzoh/A3DBaYGEj2H/HFZCn/kMfim73FXxEJTw06WtxQwg==",
"license": "MIT",
"engines": {
"node": ">= 14.18.0"
},
"funding": {
"type": "individual",
"url": "https://paulmillr.com/funding/"
}
},
"node_modules/real-require": {
"version": "0.2.0",
"resolved": "https://registry.npmjs.org/real-require/-/real-require-0.2.0.tgz",

View file

@ -60,6 +60,7 @@
"@modelcontextprotocol/sdk": "^1.0.0",
"@scarf/scarf": "^1.4.0",
"busboy": "^1.6.0",
"chokidar": "^4.0.3",
"cli-progress": "^3.12.0",
"commander": "^15.0.0",
"cors": "^2.8.5",

View file

@ -3,7 +3,7 @@
*
* WHY THIS EXISTS. `run-cross-platform.ts` used to hand vitest the whole file
* list plus `--shard=i/n`, and vitest partitions by file COUNT. Runtime on this
* suite is wildly uneven measured on the Windows runner, `cli-e2e` is 361 s
* suite is wildly uneven measured on the Windows runner, `cli-e2e` is 621 s
* and `worker-pool` 221 s, while most files are under a second so a
* count-split routinely put several of the heaviest suites on one shard. That
* is #2449, and this file's sibling header has documented the symptom ("the
@ -44,7 +44,9 @@
* partition depend on the very machine load it is trying to protect against.
*/
export const WINDOWS_WEIGHTS_SEC: Readonly<Record<string, number>> = {
'test/integration/cli-e2e.test.ts': 361,
// Re-measured after the analyze --watch e2e landed in #3072. The previous
// 361 s entry undercharged this suite and left shard 1 close to the watchdog.
'test/integration/cli-e2e.test.ts': 621,
'test/integration/worker-pool.test.ts': 222,
'test/unit/incremental-vector-extension-ordering.test.ts': 87,
// ESTIMATE, not a measurement (#2841): this suite drives more full

View file

@ -234,7 +234,7 @@ const SPAWN_CLI = [
// Cheap: measured on the Windows runner at 448 ms, 53 ms and sub-second. An
// earlier attempt to register them still turned the matrix red — not from
// their own cost, but because vitest sharded by file COUNT, so inserting any
// file re-partitioned the list and happened to cluster `cli-e2e` (361 s) with
// file re-partitioned the list and happened to cluster `cli-e2e` (621 s) with
// `cli-limit-e2e` (75 s) on one shard. The split is weight-aware now
// (`scripts/cross-platform-shard.ts`), so a cheap file can no longer move a
// heavy one.
@ -273,6 +273,7 @@ const NATIVE_ADDON_SMOKE = [
// platforms (CRLF, symlinks, permissions, temp dirs)
const FILESYSTEM = [
'test/integration/filesystem-walker.test.ts',
'test/integration/watch-filesystem.test.ts',
'test/integration/markdown-processor-crlf.test.ts',
'test/integration/ignore-and-skip-e2e.test.ts',
// Pins that the bridge pairing verdict is measured before the database is

View file

@ -21,6 +21,8 @@ Run from the project root. This parses all source files, builds the knowledge gr
| Flag | Effect |
| -------------- | ---------------------------------------------------------------- |
| `--watch` | Keep a Git repository index current with serialized refreshes |
| `--debounce <ms>` | Watch quiet period before refresh (default: 300 ms) |
| `--force` | Force full re-index even if up to date |
| `--embeddings` | Enable embedding generation for semantic search (off by default) |
| `--drop-embeddings` | Drop existing embeddings on rebuild. By default, an `analyze` without `--embeddings` preserves them. |
@ -28,6 +30,8 @@ Run from the project root. This parses all source files, builds the knowledge gr
**When to run:** First time in a project, after major code changes, or when `gitnexus://repo/{name}/context` reports the index is stale. In Claude Code, a PostToolUse hook detects staleness after `git commit` and `git merge` and notifies the agent to run `analyze` — the hook does not run analyze itself, to avoid blocking the agent for up to 120s and risking KuzuDB corruption on timeout.
Use `node .gitnexus/run.cjs analyze --watch` for a long-lived local Git repository. It performs an initial analysis, queues scanner-admitted file changes, and retries intact failed batches with bounded backoff. Watch refreshes update only the graph: they skip AGENTS.md / CLAUDE.md injection and standard skill installation, so run a one-shot `analyze` when those generated files need updating. Watch rejects one-shot or context-output flags including `--force`, embedding flags, `--skills`, `--default-branch`, `--skip-agents-md`, `--skip-skills`, `--no-stats`, `--self-commit`, `--index-only`, and `--skip-git`. It never pulls remotes. Running MCP and `serve` processes periodically check for a published replacement and reopen it without a restart. MCP checks are throttled to once every five seconds, so a tool call before the next check can briefly use the previous index.
### status — Check index freshness
```bash
@ -86,5 +90,5 @@ Lists all repositories registered in `~/.gitnexus/registry.json`. The MCP `list_
## Troubleshooting
- **"Not inside a git repository"**: Run from a directory inside a git repo
- **Index is stale after re-analyzing**: Restart Claude Code to reload the MCP server
- **Index is stale after re-analyzing**: Wait for the next MCP tool call to reopen the published index; this normally takes no more than five seconds
- **Embeddings slow**: Omit `--embeddings` (it's off by default) or set `OPENAI_API_KEY` for faster API-based embedding

View file

@ -30,6 +30,7 @@
import fs from 'node:fs';
import path from 'node:path';
import { readRepoControlFile } from '../config/repo-control-file.js';
import type { AnalyzeOptions } from './analyze-options.js';
export const GITNEXUS_RC_FILENAME = '.gitnexusrc';
@ -370,7 +371,6 @@ const normalizeLevel = (
*/
export function loadAnalyzeConfig(repoRoot: string): Partial<AnalyzeOptions> | undefined {
const filePath = path.join(repoRoot, GITNEXUS_RC_FILENAME);
let raw: string;
try {
raw = fs.readFileSync(filePath, 'utf-8');
@ -379,6 +379,25 @@ export function loadAnalyzeConfig(repoRoot: string): Partial<AnalyzeOptions> | u
throw new GitNexusRcError(`Could not read ${GITNEXUS_RC_FILENAME}: ${(err as Error).message}`);
}
return parseAnalyzeConfig(raw);
}
/** Load `.gitnexusrc` through the strict bounded reader used by watch mode. */
export async function loadAnalyzeConfigStrict(
repoRoot: string,
): Promise<Partial<AnalyzeOptions> | undefined> {
let raw: string | null;
try {
raw = await readRepoControlFile(repoRoot, GITNEXUS_RC_FILENAME);
} catch (err) {
throw new GitNexusRcError(`Could not read ${GITNEXUS_RC_FILENAME}: ${(err as Error).message}`);
}
return raw === null ? undefined : parseAnalyzeConfig(raw);
}
function parseAnalyzeConfig(rawInput: string): Partial<AnalyzeOptions> {
let raw = rawInput;
// Strip a leading UTF-8 BOM: Node's 'utf-8' decode keeps it, and JSON.parse
// then fails with a confusing "Unexpected token" on an otherwise-valid file
// (#1996 tri-review). Only one leading BOM is stripped; in-string control

View file

@ -15,6 +15,10 @@
* import cycle. `analyze.ts` re-exports the type for existing importers.
*/
export interface AnalyzeOptions {
/** Keep this repository current with serialized incremental refreshes. */
watch?: boolean;
/** Watch quiet period in milliseconds. */
debounce?: string;
force?: boolean;
repairFts?: boolean;
/**

View file

@ -363,6 +363,7 @@ interface RespawnExit {
stdout?: string;
stderr?: string;
message?: string;
forwardedSignal?: NodeJS.Signals;
}
const appendOutputTail = (tail: string, chunk: unknown): string => {
@ -395,17 +396,28 @@ const runRespawnedAnalyze = (
let stdout = '';
let stderr = '';
let settled = false;
const finish = (exit: RespawnExit): void => {
if (settled) return;
settled = true;
resolve(exit);
};
let forwardedSignal: NodeJS.Signals | undefined;
const child = spawn(process.execPath, [...args], {
stdio: ['inherit', 'pipe', 'pipe'],
windowsHide: true,
env,
});
const forwardSignal = (signal: NodeJS.Signals): void => {
forwardedSignal ??= signal;
if (child.exitCode === null && child.signalCode === null) child.kill(signal);
};
const forwardSigint = () => forwardSignal('SIGINT');
const forwardSigterm = () => forwardSignal('SIGTERM');
const finish = (exit: RespawnExit): void => {
if (settled) return;
settled = true;
process.removeListener('SIGINT', forwardSigint);
process.removeListener('SIGTERM', forwardSigterm);
resolve({ ...exit, forwardedSignal });
};
process.once('SIGINT', forwardSigint);
process.once('SIGTERM', forwardSigterm);
child.stdout?.on('data', (chunk) => {
stdout = appendOutputTail(stdout, chunk);
@ -548,7 +560,16 @@ export function parseMaxOldSpaceMb(nodeOptions: string): number | null {
* tooling), not a deliberate per-run choice: warn and respawn with the
* auto cap. Pre-#2649 this returned early and large repos then OOM'd on
* whatever heap the environment happened to specify. */
async function ensureHeap(): Promise<boolean> {
export function forwardedSignalExitCode(signal: NodeJS.Signals, cleanTermination: boolean): number {
if (cleanTermination) return 0;
if (signal === 'SIGINT') return 130;
if (signal === 'SIGTERM') return 143;
return 1;
}
export async function ensureHeap(
options: { cleanForwardedTermination?: boolean } = {},
): Promise<boolean> {
// Explicit opt-out disables auto-sizing ENTIRELY — both the ambient-pin
// override and the default v8-limit respawn — and is honored SILENTLY:
// the operator already made the call, and stderr-sensitive consumers
@ -590,6 +611,13 @@ async function ensureHeap(): Promise<boolean> {
};
if (shouldBridgeRespawnProgressTty()) childEnv[RESPAWN_PROGRESS_ENV] = '1';
const childExit = await runRespawnedAnalyze(childArgs, childEnv);
if (childExit.forwardedSignal !== undefined) {
process.exitCode = forwardedSignalExitCode(
childExit.forwardedSignal,
options.cleanForwardedTermination === true,
);
return true;
}
if (childExit.status !== 0 || childExit.signal) {
if (childProcessLikelyOom(childExit)) {
cliError(
@ -740,6 +768,19 @@ export const analyzeCommandWithRunnerIdentity = async (
options?: AnalyzeOptions,
): Promise<void> => analyzeCommand(inputPath, options, runnerIdentityAtBootstrap);
export async function analyzeOrWatchCommandWithRunnerIdentity(
runnerIdentityAtBootstrap: AnalyzerRunnerIdentity,
inputPath?: string,
options: AnalyzeOptions = {},
): Promise<void> {
if (options.watch) {
const { watchCommandWithRunnerIdentity } = await import('./watch.js');
await watchCommandWithRunnerIdentity(runnerIdentityAtBootstrap, inputPath, options);
return;
}
await analyzeCommandWithRunnerIdentity(runnerIdentityAtBootstrap, inputPath, options);
}
const analyzeCommandImpl = async (
inputPath?: string,
cliOptions?: AnalyzeOptions,

View file

@ -72,6 +72,8 @@ const OPTION_DESCRIPTION_KEYS = {
'analyze|--embedding-batch-size <n>': 'help.option.analyze.embeddingBatchSize',
'analyze|--embedding-sub-batch-size <n>': 'help.option.analyze.embeddingSubBatchSize',
'analyze|--embedding-device <device>': 'help.option.analyze.embeddingDevice',
'analyze|--watch': 'help.option.analyze.watch',
'analyze|--debounce <ms>': 'help.option.analyze.debounce',
'index|-f, --force': 'help.option.index.force',
'index|--allow-non-git': 'help.option.index.allowNonGit',
'mcp|--http': 'help.option.mcp.http',

View file

@ -217,6 +217,8 @@ export const en = {
'help.option.analyze.embeddingBatchSize': 'Number of nodes per embedding batch',
'help.option.analyze.embeddingSubBatchSize': 'Number of chunks per embedding model call',
'help.option.analyze.embeddingDevice': 'Embedding device: auto, cpu, dml, cuda, or wasm',
'help.option.analyze.watch': 'Keep the index current with serialized incremental refreshes',
'help.option.analyze.debounce': 'Watch quiet period before refreshing (milliseconds)',
'help.option.index.force': 'Register even if index metadata is missing (stats will be empty)',
'help.option.index.allowNonGit': 'Allow registering folders that are not Git repositories',
'help.option.port': 'Port number',

View file

@ -203,6 +203,8 @@ export const zhCN = {
'help.option.analyze.embeddingBatchSize': '每个嵌入批次的节点数',
'help.option.analyze.embeddingSubBatchSize': '每次嵌入模型调用的分块数',
'help.option.analyze.embeddingDevice': '嵌入设备auto、cpu、dml、cuda 或 wasm',
'help.option.analyze.watch': '监视本地源文件变更并串行执行增量刷新',
'help.option.analyze.debounce': '刷新前的静默等待时间(毫秒)',
'help.option.index.force': '即使缺少索引元数据也注册(统计为空)',
'help.option.index.allowNonGit': '允许注册非 Git 仓库文件夹',
'help.option.port': '端口号',

View file

@ -57,6 +57,8 @@ let dimsEnvCaptured = false;
program
.command('analyze [path]')
.description('Index a repository (full analysis)')
.option('--watch', 'Keep the index current with serialized incremental refreshes')
.option('--debounce <ms>', 'Watch quiet period before refreshing (default: 300 milliseconds)')
.option('-f, --force', 'Force full re-index even if up to date')
.option('--repair-fts', 'Repair/rebuild search FTS indexes without full re-analysis')
.option(
@ -162,6 +164,11 @@ program
)
.addHelpText('after', () => t('help.analyze.environment'))
.hook('preAction', (thisCommand: Command) => {
const analyzeOpts = thisCommand.opts();
if (analyzeOpts['debounce'] !== undefined && analyzeOpts['watch'] !== true) {
process.stderr.write('\n --debounce requires --watch\n\n');
process.exit(1);
}
// ONLY GITNEXUS_EMBEDDING_DIMS must be set here: schema.ts reads it at
// module-load time during the lazy import('./analyze.js') below (via the
// static chain analyze.ts → run-analyze.ts → schema.ts), so deferring to
@ -169,7 +176,7 @@ program
// lazily at runtime (readConfig), so analyzeCommandImpl is their sole
// setter — keeping them out of this hook means they fall under the impl's
// env snapshot/restore and don't leak across in-process invocations.
const dimsOpt = thisCommand.opts()['embeddingDims'];
const dimsOpt = analyzeOpts['embeddingDims'];
if (dimsOpt !== undefined) {
// Validate + normalize BEFORE writing the env var: schema.ts throws on a
// bad value at module-load, which — on the synchronous program.parse()
@ -202,7 +209,7 @@ program
createAnalyzerLbugLazyAction(
() => import('../core/analyzer-identity.js'),
() => import('./analyze.js'),
'analyzeCommandWithRunnerIdentity',
'analyzeOrWatchCommandWithRunnerIdentity',
import.meta.url,
),
);

View file

@ -0,0 +1,184 @@
export type WatchRefresh = (paths: readonly string[]) => Promise<void>;
export type WatchRefreshError = (error: unknown, paths: readonly string[]) => void;
export const WATCH_FULL_REFRESH_PATH = '*';
export interface WatchRefreshQueueOptions {
readonly maxWaitMs?: number;
readonly maxPendingPaths?: number;
readonly retryBaseDelayMs?: number;
readonly retryMaxDelayMs?: number;
readonly holdEventsUntilInitialRefresh?: boolean;
readonly isPriorityPath?: (filePath: string) => boolean;
}
/** Debounces filesystem events and guarantees that refreshes never overlap. */
export class WatchRefreshQueue {
private readonly pending = new Set<string>();
private readonly idleWaiters = new Set<() => void>();
private timer: ReturnType<typeof setTimeout> | undefined;
private active: Promise<void> | undefined;
private closed = false;
private initialPending = false;
private firstPendingAt: number | undefined;
private overflowed = false;
private consecutiveFailures = 0;
private retryNotBefore: number | undefined;
constructor(
private readonly refresh: WatchRefresh,
private readonly onError: WatchRefreshError,
private readonly debounceMs: number,
private readonly options: WatchRefreshQueueOptions = {},
) {
this.initialPending = options.holdEventsUntilInitialRefresh === true;
}
enqueue(filePath: string): void {
if (this.closed) return;
this.addPendingPath(filePath);
this.firstPendingAt ??= Date.now();
if (!this.initialPending && this.active === undefined) this.schedule();
}
private addPendingPath(filePath: string): void {
const maxPendingPaths = this.options.maxPendingPaths ?? 1_000;
const priority = this.options.isPriorityPath?.(filePath) === true;
if (this.pending.has(filePath)) {
// A duplicate does not increase memory use or imply that paths were dropped.
} else if (this.pending.size < maxPendingPaths) {
this.pending.add(filePath);
} else {
this.overflowed = true;
if (priority) {
const evictable = [...this.pending].find(
(pendingPath) => this.options.isPriorityPath?.(pendingPath) !== true,
);
if (evictable !== undefined) {
this.pending.delete(evictable);
this.pending.add(filePath);
}
}
}
}
/** Run the initial refresh while still queueing events that arrive during it. */
async runInitial(): Promise<void> {
if (this.closed) return;
if (this.active !== undefined) throw new Error('Watch refresh is already running');
try {
await this.runBatch([], true);
} finally {
this.initialPending = false;
if (!this.closed && this.hasPendingWork()) this.schedule();
else this.resolveIdleWaiters();
}
}
async waitForIdle(): Promise<void> {
if (this.isIdle()) return;
await new Promise<void>((resolve) => this.idleWaiters.add(resolve));
}
async close(): Promise<void> {
this.closed = true;
if (this.timer !== undefined) clearTimeout(this.timer);
this.timer = undefined;
this.pending.clear();
this.firstPendingAt = undefined;
this.overflowed = false;
this.consecutiveFailures = 0;
this.retryNotBefore = undefined;
// A refresh rejection is already surfaced through `onError` (or through
// runInitial). Closing from that handler can race the runBatch `finally`,
// so consume the same rejection here instead of reporting it twice.
await this.active?.catch(() => {});
this.resolveIdleWaiters();
}
private schedule(retryDelayMs?: number): void {
if (this.timer !== undefined) clearTimeout(this.timer);
const maxWaitMs = this.options.maxWaitMs ?? Math.max(this.debounceMs, 2_000);
const now = Date.now();
if (retryDelayMs !== undefined) this.retryNotBefore = now + retryDelayMs;
const elapsed = this.firstPendingAt === undefined ? 0 : now - this.firstPendingAt;
const debounced = Math.max(0, Math.min(this.debounceMs, maxWaitMs - elapsed));
// An event arriving mid-backoff merges into the pending batch but must not
// pull the retry earlier than the deadline the backoff already committed to.
const delay =
retryDelayMs ??
(this.retryNotBefore === undefined
? debounced
: Math.max(debounced, this.retryNotBefore - now));
this.timer = setTimeout(() => {
this.timer = undefined;
void this.drain();
}, delay);
}
private async drain(): Promise<void> {
if (this.closed || this.active !== undefined || !this.hasPendingWork()) return;
const paths = [
...(this.overflowed ? [WATCH_FULL_REFRESH_PATH] : []),
...[...this.pending].sort(),
];
this.pending.clear();
this.firstPendingAt = undefined;
this.overflowed = false;
this.retryNotBefore = undefined;
await this.runBatch(paths, false);
}
private async runBatch(paths: readonly string[], propagateError: boolean): Promise<void> {
let work: Promise<void>;
try {
work = this.refresh(paths);
} catch (error) {
work = Promise.reject(error);
}
this.active = work;
let retryDelayMs: number | undefined;
try {
await work;
this.consecutiveFailures = 0;
} catch (error) {
if (propagateError) throw error;
try {
await this.onError(error, paths);
} catch {
// Refresh failures are already handled here; a reporter must not
// reject the detached drain promise and become an unhandled rejection.
}
if (!this.closed) {
if (paths.includes(WATCH_FULL_REFRESH_PATH)) this.overflowed = true;
for (const filePath of paths) {
if (filePath !== WATCH_FULL_REFRESH_PATH) this.addPendingPath(filePath);
}
this.firstPendingAt = Date.now();
this.consecutiveFailures++;
const base = this.options.retryBaseDelayMs ?? Math.max(250, this.debounceMs);
const maximum = this.options.retryMaxDelayMs ?? 30_000;
retryDelayMs = Math.min(maximum, base * 2 ** (this.consecutiveFailures - 1));
}
} finally {
if (this.active === work) this.active = undefined;
if (!this.closed && !this.initialPending && this.hasPendingWork())
this.schedule(retryDelayMs);
else this.resolveIdleWaiters();
}
}
private hasPendingWork(): boolean {
return this.overflowed || this.pending.size > 0;
}
private isIdle(): boolean {
return this.active === undefined && this.timer === undefined && !this.hasPendingWork();
}
private resolveIdleWaiters(): void {
if (!this.isIdle() && !this.closed) return;
for (const resolve of this.idleWaiters) resolve();
this.idleWaiters.clear();
}
}

501
gitnexus/src/cli/watch.ts Normal file
View file

@ -0,0 +1,501 @@
import path from 'node:path';
import fs from 'node:fs/promises';
import { watch, type FSWatcher } from 'chokidar';
import { createWatchIgnorePredicate } from '../config/ignore-service.js';
import {
analyzeFailureMayHaveMutatedLiveIndex,
runFullAnalysis,
type AnalyzeOptions as CoreAnalyzeOptions,
type AnalyzeResult,
} from '../core/run-analyze.js';
import { getGitRoot, hasGitDir } from '../storage/git.js';
import type { AnalyzerRunnerIdentity } from '../storage/repo-manager.js';
import { GITNEXUS_DIR } from '../storage/repo-meta.js';
import {
loadAnalyzeConfigStrict,
mergeAnalyzeOptions,
validateBranchName,
} from './analyze-config.js';
import type { AnalyzeOptions } from './analyze-options.js';
import { ensureHeap } from './analyze.js';
import { cliError, cliInfo, cliWarn } from './cli-message.js';
import {
WATCH_FULL_REFRESH_PATH,
WatchRefreshQueue,
type WatchRefreshError,
} from './watch-queue.js';
const DEFAULT_DEBOUNCE_MS = 300;
const MAX_TIMER_DELAY_MS = 2_147_483_647;
const MAX_FILE_SIZE_KB = 32 * 1024;
const TRANSIENT_WATCH_ERROR_CODES = new Set(['EACCES', 'ENOENT', 'ENOTDIR', 'EPERM']);
export type WatchCliOptions = AnalyzeOptions;
function posixWatchPath(filePath: string): string {
return filePath.replace(/\\/g, '/').replace(/^\.\/+/, '');
}
export function isRelevantWatchPath(filePath: string): boolean {
const normalized = posixWatchPath(filePath);
return (
normalized.length > 0 &&
normalized !== '.' &&
!normalized.startsWith('../') &&
!path.posix.isAbsolute(normalized) &&
!path.win32.isAbsolute(filePath)
);
}
function isIgnoreControlPath(filePath: string): boolean {
const normalized = posixWatchPath(filePath);
return normalized === '.gitignore' || normalized === '.gitnexusignore';
}
function isConfigControlPath(filePath: string): boolean {
return posixWatchPath(filePath) === '.gitnexusrc';
}
function isAnalyzerOwnedWatchPath(filePath: string): boolean {
const normalized = posixWatchPath(filePath).replace(/\/+$/, '');
return normalized === GITNEXUS_DIR || normalized.startsWith(`${GITNEXUS_DIR}/`);
}
function repoRelativeWatchPath(repoPath: string, candidate: string): string | null {
const relative = path.relative(repoPath, candidate).replace(/\\/g, '/');
if (!relative || relative.startsWith('../') || path.isAbsolute(relative)) return null;
return relative;
}
export interface WatchEnvironmentBaseline {
readonly maxFileSize: string | undefined;
readonly workerTimeout: string | undefined;
readonly verbose: string | undefined;
}
function setEnvironment(name: string, value: string | undefined): void {
if (value === undefined) delete process.env[name];
else process.env[name] = value;
}
function positiveInteger(
value: string | undefined,
flag: string,
maximum?: number,
): number | undefined {
if (value === undefined) return undefined;
const parsed = Number(value);
if (!Number.isInteger(parsed) || parsed < 1)
throw new Error(`${flag} must be a positive integer`);
if (maximum !== undefined && parsed > maximum) {
throw new Error(`${flag} must not exceed ${maximum}`);
}
return parsed;
}
export async function resolveWatchOptions(
repoPath: string,
cli: WatchCliOptions,
baseline: WatchEnvironmentBaseline,
reportIgnoredConfig: (names: readonly string[]) => void = () => {},
): Promise<CoreAnalyzeOptions> {
const config = (await loadAnalyzeConfigStrict(repoPath)) ?? {};
const merged = mergeAnalyzeOptions(cli, config);
const unsupported = [
['--force', cli.force],
['--repair-fts', cli.repairFts],
['--embeddings', cli.embeddings],
['--drop-embeddings', cli.dropEmbeddings],
['--skills', cli.skills],
['--default-branch', cli.defaultBranch],
['--skip-agents-md', cli.skipAgentsMd],
['--skip-skills', cli.skipSkills],
['--no-stats', cli.stats === false],
['--self-commit', cli.selfCommit],
['--index-only', cli.indexOnly],
['--skip-git', cli.skipGit],
['walCheckpointThreshold', cli.walCheckpointThreshold],
['embeddingThreads', cli.embeddingThreads],
['embeddingBatchSize', cli.embeddingBatchSize],
['embeddingSubBatchSize', cli.embeddingSubBatchSize],
['embeddingDevice', cli.embeddingDevice],
['embeddingBaseUrl', cli.embeddingBaseUrl],
['embeddingModel', cli.embeddingModel],
['--embedding-auth-token', cli.embeddingAuthToken],
['--embedding-dims', cli.embeddingDims],
].filter(([, value]) => value !== undefined && value !== false);
if (unsupported.length > 0) {
throw new Error(
`analyze --watch does not support ${unsupported.map(([name]) => name).join(', ')}`,
);
}
reportIgnoredConfig(
[
['embeddings', config.embeddings],
['dropEmbeddings', config.dropEmbeddings],
['defaultBranch', config.defaultBranch],
['skipAgentsMd', config.skipAgentsMd !== undefined],
['skipSkills', config.skipSkills !== undefined],
['stats', config.stats !== undefined],
['walCheckpointThreshold', config.walCheckpointThreshold],
['embeddingThreads', config.embeddingThreads],
['embeddingBatchSize', config.embeddingBatchSize],
['embeddingSubBatchSize', config.embeddingSubBatchSize],
['embeddingDevice', config.embeddingDevice],
['embeddingBaseUrl', config.embeddingBaseUrl],
['embeddingModel', config.embeddingModel],
]
.filter(([, value]) => value !== undefined && value !== false)
.map(([name]) => String(name)),
);
const branch =
merged.branch === undefined ? undefined : validateBranchName(merged.branch, '--branch');
const workerPoolSize = positiveInteger(merged.workers, '--workers');
const workerTimeoutSeconds = positiveInteger(merged.workerTimeout, 'workerTimeout');
const maxFileSize = positiveInteger(merged.maxFileSize, 'maxFileSize', MAX_FILE_SIZE_KB);
setEnvironment(
'GITNEXUS_MAX_FILE_SIZE',
maxFileSize === undefined ? baseline.maxFileSize : String(maxFileSize),
);
if (workerTimeoutSeconds !== undefined) {
process.env.GITNEXUS_WORKER_SUB_BATCH_TIMEOUT_MS = String(workerTimeoutSeconds * 1000);
} else {
setEnvironment('GITNEXUS_WORKER_SUB_BATCH_TIMEOUT_MS', baseline.workerTimeout);
}
setEnvironment('GITNEXUS_VERBOSE', merged.verbose ? '1' : baseline.verbose);
return {
pdg: merged.pdg,
branch,
registryName: merged.name,
allowDuplicateName: merged.allowDuplicateName,
workerPoolSize,
fetchWrappers: merged.fetchWrappers,
skipAgentsMd: true,
skipSkills: true,
noStats: true,
atomicIncremental: process.platform !== 'win32',
};
}
function refreshSummary(
result: AnalyzeResult,
observedPaths: readonly string[],
durationMs: number,
lastSuccessfulRefreshAt: string,
): string {
const measured = result.incrementalStats;
const changed = measured?.changedFiles ?? (result.alreadyUpToDate ? 0 : observedPaths.length);
const reparsed =
measured?.reparsedFiles ??
(typeof result.pipelineResult?.reparsedFileCount === 'number'
? result.pipelineResult.reparsedFileCount
: 0);
const dependents = measured?.affectedDependents ?? 0;
const mode = measured?.writeMode ?? (result.alreadyUpToDate ? 'no-op' : 'full');
return (
`Refresh complete: ${changed} changed, ${reparsed} re-parsed, ` +
`${dependents} affected dependent(s), ${durationMs}ms, ${mode}; ` +
`last success ${lastSuccessfulRefreshAt}`
);
}
async function waitUntilReady(watcher: FSWatcher): Promise<void> {
await new Promise<void>((resolve, reject) => {
const ready = () => {
watcher.off('error', failed);
resolve();
};
const failed = (error: unknown) => {
watcher.off('ready', ready);
reject(error);
};
watcher.once('ready', ready);
watcher.once('error', failed);
});
}
export interface WatchFileLoop {
readonly waitForIdle: () => Promise<void>;
readonly close: () => Promise<void>;
}
class WatchControlReloadError extends Error {
constructor(cause: unknown) {
super(cause instanceof Error ? cause.message : String(cause), { cause });
this.name = 'WatchControlReloadError';
}
}
export function shouldStopAfterWatchRefreshFailure(
error: unknown,
paths: readonly string[],
): boolean {
return (
paths.length > 0 &&
!(error instanceof WatchControlReloadError) &&
analyzeFailureMayHaveMutatedLiveIndex(error)
);
}
/** Start the real filesystem watcher with bounded, serialized refreshes. */
export async function startWatchFileLoop(
repoPath: string,
debounceMs: number,
refresh: (paths: readonly string[]) => Promise<void>,
onError: WatchRefreshError,
onWatcherError: (error: unknown) => void = (error) => onError(error, []),
): Promise<WatchFileLoop> {
let ignorePath = await createWatchIgnorePredicate(repoPath);
let ignoreControlValid = true;
const queue = new WatchRefreshQueue(
async (paths) => {
if (paths.some(isIgnoreControlPath) || !ignoreControlValid) {
const retryingInvalidControls = !ignoreControlValid;
try {
ignorePath = await createWatchIgnorePredicate(repoPath);
ignoreControlValid = true;
watcher.add(repoPath);
} catch (error) {
ignoreControlValid = false;
throw new WatchControlReloadError(
retryingInvalidControls
? new Error(
'Ignore controls remain invalid; fix them before indexing more changes.',
{
cause: error,
},
)
: error,
);
}
}
await refresh(paths);
},
onError,
debounceMs,
{
maxWaitMs: Math.max(2_000, debounceMs * 10),
maxPendingPaths: 1_000,
holdEventsUntilInitialRefresh: true,
isPriorityPath: (filePath) => isIgnoreControlPath(filePath) || isConfigControlPath(filePath),
},
);
const watcher: FSWatcher = watch(repoPath, {
ignoreInitial: true,
atomic: true,
followSymlinks: false,
awaitWriteFinish: { stabilityThreshold: 100, pollInterval: 20 },
ignored: (candidate, stats) => {
const relative = repoRelativeWatchPath(repoPath, candidate);
if (relative !== null && isAnalyzerOwnedWatchPath(relative)) return true;
if (relative !== null && (isIgnoreControlPath(relative) || isConfigControlPath(relative))) {
return false;
}
return ignorePath(candidate, stats?.isDirectory() ?? false);
},
});
watcher.on('all', (event, changedPath) => {
if (event !== 'add' && event !== 'change' && event !== 'unlink') return;
const relative = repoRelativeWatchPath(repoPath, changedPath);
if (relative && isRelevantWatchPath(relative) && !isAnalyzerOwnedWatchPath(relative)) {
queue.enqueue(relative);
}
});
watcher.on('error', (error) => {
// Chokidar can surface a transient EPERM on Windows while an ignored
// analyzer-owned path is replaced. Re-arm the root and force one bounded
// catch-up refresh so a missed event cannot leave the graph stale. Other
// watcher errors may mean coverage was lost and remain fatal.
if (TRANSIENT_WATCH_ERROR_CODES.has((error as NodeJS.ErrnoException).code ?? '')) {
watcher.add(repoPath);
queue.enqueue(WATCH_FULL_REFRESH_PATH);
return;
}
onWatcherError(error);
});
try {
await waitUntilReady(watcher);
await queue.runInitial();
} catch (error) {
await watcher.close();
await queue.close();
throw error;
}
return {
waitForIdle: () => queue.waitForIdle(),
close: async () => {
await watcher.close();
await queue.close();
},
};
}
export async function watchCommandWithRunnerIdentity(
runnerIdentityAtBootstrap: AnalyzerRunnerIdentity,
inputPath?: string,
cliOptions: WatchCliOptions = {},
): Promise<void> {
if (await ensureHeap({ cleanForwardedTermination: true })) return;
const requestedRepoPath = inputPath ? path.resolve(inputPath) : getGitRoot(process.cwd());
if (requestedRepoPath === null || !hasGitDir(requestedRepoPath)) {
cliError(' gitnexus analyze --watch requires a Git repository.');
process.exitCode = 1;
return;
}
const repoPath = await fs.realpath(requestedRepoPath);
const baselineEnvironment: WatchEnvironmentBaseline = {
maxFileSize: process.env.GITNEXUS_MAX_FILE_SIZE,
workerTimeout: process.env.GITNEXUS_WORKER_SUB_BATCH_TIMEOUT_MS,
verbose: process.env.GITNEXUS_VERBOSE,
};
try {
let ignoredConfigSignature: string | undefined;
const reportIgnoredConfig = (names: readonly string[]) => {
const signature = [...names].sort().join(',');
if (signature === ignoredConfigSignature) return;
ignoredConfigSignature = signature;
if (names.length > 0) {
cliWarn(`Watch mode ignores unsupported .gitnexusrc settings: ${names.join(', ')}.`);
}
};
let debounceMs: number;
let analyzeOptions: CoreAnalyzeOptions;
try {
debounceMs =
positiveInteger(
cliOptions.debounce ?? String(DEFAULT_DEBOUNCE_MS),
'--debounce',
MAX_TIMER_DELAY_MS,
) ?? DEFAULT_DEBOUNCE_MS;
analyzeOptions = await resolveWatchOptions(
repoPath,
cliOptions,
baselineEnvironment,
reportIgnoredConfig,
);
} catch (error) {
cliError(` ${error instanceof Error ? error.message : String(error)}`);
process.exitCode = 1;
return;
}
let stopWatching!: () => void;
const stopped = new Promise<void>((resolve) => {
stopWatching = resolve;
});
const stop = () => stopWatching();
process.once('SIGINT', stop);
process.once('SIGTERM', stop);
try {
let loop: WatchFileLoop;
let fatalRefreshError: unknown;
let configControlValid = true;
let lastSuccessfulRefreshAt: string | undefined;
try {
loop = await startWatchFileLoop(
repoPath,
debounceMs,
async (paths) => {
if (paths.some(isConfigControlPath) || !configControlValid) {
const retryingInvalidConfig = !configControlValid;
try {
analyzeOptions = await resolveWatchOptions(
repoPath,
cliOptions,
baselineEnvironment,
reportIgnoredConfig,
);
configControlValid = true;
} catch (error) {
configControlValid = false;
throw new WatchControlReloadError(
retryingInvalidConfig
? new Error(
'Configuration remains invalid; fix it before indexing more changes.',
{
cause: error,
},
)
: error,
);
}
}
const startedAt = Date.now();
const result = await runFullAnalysis(
repoPath,
analyzeOptions,
{
onProgress: () => {},
onLog:
process.env.GITNEXUS_VERBOSE === '1'
? (message) => cliInfo(` ${message}`)
: undefined,
},
runnerIdentityAtBootstrap,
);
lastSuccessfulRefreshAt = new Date().toISOString();
if (paths.length === 0) {
cliInfo(
result.alreadyUpToDate
? `Watching ${repoPath}; index is up to date.`
: `Watching ${repoPath}; initial index ready in ${Date.now() - startedAt}ms.`,
);
} else {
cliInfo(
refreshSummary(result, paths, Date.now() - startedAt, lastSuccessfulRefreshAt),
);
}
},
(error, paths) => {
const detail = paths.length > 0 ? ` (${paths.length} queued path(s))` : '';
if (shouldStopAfterWatchRefreshFailure(error, paths)) {
fatalRefreshError = error;
cliError(
`Refresh failed${detail}: ${error instanceof Error ? error.message : String(error)}. ` +
'Watch mode is stopping because the live index may have been updated in place.',
);
stopWatching();
return;
}
const lastSuccess = lastSuccessfulRefreshAt ?? 'none yet';
cliWarn(
`Refresh failed${detail}: ${error instanceof Error ? error.message : String(error)}. ` +
`Retry scheduled; last success ${lastSuccess}.`,
);
},
(error) => {
fatalRefreshError = error;
cliError(
`Watcher failed: ${error instanceof Error ? error.message : String(error)}. ` +
'Watch mode is stopping.',
);
stopWatching();
},
);
} catch (error) {
cliError(
` Unable to start watcher: ${error instanceof Error ? error.message : String(error)}`,
);
process.exitCode = 1;
return;
}
await stopped;
await loop.close();
if (fatalRefreshError !== undefined) process.exitCode = 1;
} finally {
process.removeListener('SIGINT', stop);
process.removeListener('SIGTERM', stop);
}
} finally {
setEnvironment('GITNEXUS_MAX_FILE_SIZE', baselineEnvironment.maxFileSize);
setEnvironment('GITNEXUS_WORKER_SUB_BATCH_TIMEOUT_MS', baselineEnvironment.workerTimeout);
setEnvironment('GITNEXUS_VERBOSE', baselineEnvironment.verbose);
}
}

View file

@ -3,6 +3,7 @@ import { existsSync } from 'fs';
import fs from 'fs/promises';
import nodePath from 'path';
import type { Path } from 'path-scurry';
import { readRepoControlFile } from './repo-control-file.js';
import { logger } from '../core/logger.js';
import { getCoreExcludesFilePath, getGitInfoExcludePath } from '../storage/git.js';
@ -401,6 +402,8 @@ export interface IgnoreOptions {
noGitignore?: boolean;
/** Skip core.excludesFile and $GIT_COMMON_DIR/info/exclude. Defaults to GITNEXUS_NO_GLOBAL_IGNORE env var. */
noGlobalIgnore?: boolean;
/** Fail repository-control reloads closed so long-lived watchers keep their prior predicate. */
strictRepoControlFiles?: boolean;
}
export const loadIgnoreRules = async (
@ -442,20 +445,56 @@ export const loadIgnoreRules = async (
for (const filename of filenames) {
try {
const content = await fs.readFile(nodePath.join(repoPath, filename), 'utf-8');
const content = options?.strictRepoControlFiles
? await readRepoControlFile(repoPath, filename)
: await fs.readFile(nodePath.join(repoPath, filename), 'utf-8');
if (content === null) continue;
ig.add(content);
hasRules = true;
} catch (err: unknown) {
const code = (err as NodeJS.ErrnoException).code;
if (code !== 'ENOENT') {
logger.warn(` Warning: could not read ${filename}: ${(err as Error).message}`);
}
if (!options?.strictRepoControlFiles && code === 'ENOENT') continue;
if (options?.strictRepoControlFiles) throw err;
logger.warn(` Warning: could not read ${filename}: ${(err as Error).message}`);
}
}
return hasRules ? ig : null;
};
/**
* Build a synchronous predicate for long-lived filesystem watchers.
*
* Unlike {@link createIgnoreFilter}, callers pass ordinary absolute or
* repository-relative paths instead of path-scurry `Path` objects. The rule
* precedence deliberately mirrors the scanner: explicit negations win over
* hardcoded defaults unless a more-specific rule re-ignores the path.
*/
export const createWatchIgnorePredicate = async (
repoPath: string,
options?: IgnoreOptions,
): Promise<(candidatePath: string, isDirectory?: boolean) => boolean> => {
const ig = await loadIgnoreRules(repoPath, { ...options, strictRepoControlFiles: true });
const repoRoot = nodePath.resolve(repoPath);
return (candidatePath: string, isDirectory = false): boolean => {
const absolute = nodePath.isAbsolute(candidatePath)
? nodePath.resolve(candidatePath)
: nodePath.resolve(repoRoot, candidatePath);
const rel = nodePath.relative(repoRoot, absolute).replace(/\\/g, '/');
if (!rel) return false;
if (rel === '..' || rel.startsWith('../') || nodePath.isAbsolute(rel)) return true;
if (ig && hasExplicitUnignore(ig, rel) && !ig.ignores(isDirectory ? `${rel}/` : rel)) {
return false;
}
if (ig && ig.ignores(isDirectory ? `${rel}/` : rel)) return true;
if (isDirectory && isHardcodedIgnoredDirectoryAtPath(repoRoot, absolute)) return true;
return shouldIgnorePath(rel);
};
};
/**
* Walk ancestor segments of `rel` and check whether `.gitnexusignore`
* (or `.gitignore`) contains an explicit `!pattern` negation that

View file

@ -0,0 +1,117 @@
import fs from 'node:fs';
import * as path from 'node:path';
export const MAX_REPO_CONTROL_FILE_BYTES = 1024 * 1024;
/** Read a bounded, regular control file owned by the repository root. */
export async function readRepoControlFile(
repoRoot: string,
filename: string,
): Promise<string | null> {
const requestedRoot = path.resolve(repoRoot);
const requested = path.resolve(requestedRoot, filename);
const relative = path.relative(requestedRoot, requested);
if (relative.startsWith('..') || path.isAbsolute(relative)) {
throw new Error(`${filename} resolves outside the repository root`);
}
try {
const canonicalRoot = fs.realpathSync(requestedRoot);
const beforeOpen = fs.lstatSync(requested);
if (beforeOpen.isSymbolicLink()) throw new Error(`${filename} must not be a symbolic link`);
if (!beforeOpen.isFile()) throw new Error(`${filename} must be a regular file`);
if (beforeOpen.nlink !== 1) throw new Error(`${filename} must not be a hard link`);
if (beforeOpen.size > MAX_REPO_CONTROL_FILE_BYTES) {
throw new Error(`${filename} exceeds ${MAX_REPO_CONTROL_FILE_BYTES} bytes`);
}
return await new Promise<string>((resolve, reject) => {
const stream = fs.createReadStream(requested, {
flags: 'r',
start: 0,
end: MAX_REPO_CONTROL_FILE_BYTES,
autoClose: true,
});
const chunks: Buffer[] = [];
let totalBytes = 0;
let validated = false;
let settled = false;
const finish = (value: string): void => {
if (settled) return;
settled = true;
resolve(value);
};
const fail = (error: unknown): void => {
if (settled) return;
settled = true;
reject(error);
};
stream.pause();
stream.once('open', (fd) => {
try {
const opened = fs.fstatSync(fd);
if (!opened.isFile()) throw new Error(`${filename} must be a regular file`);
if (opened.nlink !== 1) throw new Error(`${filename} must not be a hard link`);
if (opened.size > MAX_REPO_CONTROL_FILE_BYTES) {
throw new Error(`${filename} exceeds ${MAX_REPO_CONTROL_FILE_BYTES} bytes`);
}
const entry = fs.lstatSync(requested);
if (entry.isSymbolicLink()) throw new Error(`${filename} must not be a symbolic link`);
if (
!entry.isFile() ||
entry.nlink !== 1 ||
entry.dev !== opened.dev ||
entry.ino !== opened.ino
) {
throw new Error(`${filename} moved or was replaced while being opened`);
}
const canonicalFile = fs.realpathSync(requested);
const canonicalRelative = path.relative(canonicalRoot, canonicalFile);
if (canonicalRelative.startsWith('..') || path.isAbsolute(canonicalRelative)) {
throw new Error(`${filename} resolves outside the repository root`);
}
const canonical = fs.statSync(canonicalFile);
if (
canonical.nlink !== 1 ||
canonical.dev !== opened.dev ||
canonical.ino !== opened.ino
) {
throw new Error(`${filename} moved or was replaced while being opened`);
}
validated = true;
stream.resume();
} catch (error) {
fail(error);
stream.destroy();
}
});
stream.on('data', (chunk: Buffer | string) => {
const bytes = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk);
totalBytes += bytes.length;
if (totalBytes > MAX_REPO_CONTROL_FILE_BYTES) {
fail(new Error(`${filename} exceeds ${MAX_REPO_CONTROL_FILE_BYTES} bytes`));
stream.destroy();
return;
}
chunks.push(bytes);
});
stream.once('end', () => {
if (!validated) {
fail(new Error(`${filename} could not be validated`));
return;
}
finish(Buffer.concat(chunks, totalBytes).toString('utf8'));
});
stream.once('error', fail);
stream.once('close', () => {
if (!settled) fail(new Error(`${filename} closed before it could be read`));
});
});
} catch (error) {
if ((error as NodeJS.ErrnoException).code === 'ENOENT') return null;
throw error;
}
}

View file

@ -477,6 +477,8 @@ export async function runChunkedParseAndResolve(
* files. There is no sequential parser the pool is the sole parse path
* whenever a chunk misses the cache. */
usedWorkerPool: boolean;
/** Files dispatched to parser workers after parse-cache lookup. */
reparsedFileCount: number;
/** Worker-produced ParsedFile artifacts aggregated across chunks.
* Threaded into scope-resolution as a re-extract cache so the warm-
* cache analyze run can skip the dominant `extractParsedFile` cost
@ -783,6 +785,7 @@ export async function runChunkedParseAndResolve(
: new Set<string>();
let chunkCacheHits = 0;
let chunkCacheMisses = 0;
let reparsedFileCount = 0;
try {
// U1 — bounded chunk concurrency (B1 from PR #1693 review): pre-fetch
@ -1106,6 +1109,7 @@ export async function runChunkedParseAndResolve(
// Cache miss: dispatch to workers, capture the raw results, store
// them under the chunk hash for the next run.
chunkCacheMisses++;
reparsedFileCount += chunkFiles.length;
if (durableParsedFileDir !== undefined && chunkHash !== null) {
try {
await prepareDurableParsedFileChunk(durableParsedFileDir, chunkHash);
@ -1622,6 +1626,11 @@ export async function runChunkedParseAndResolve(
// no pool was needed: a warm all-cache-hit run replays cached worker output
// without spawning workers, or there were no parseable files.
usedWorkerPool: workerPool !== undefined,
// Exact number of files sent through workers on parse-cache misses. A
// changed file can invalidate its whole content-addressed chunk, so this
// is intentionally measured at dispatch time rather than inferred from
// the git/hash diff.
reparsedFileCount,
// Per-file ParsedFile artifacts produced by workers' calls to
// `extractParsedFile`. Consumed by scope-resolution as a re-extraction
// cache: when the file's ParsedFile is here, scope-resolution skips its own

View file

@ -71,6 +71,8 @@ export interface ParseOutput {
* is no sequential parser; the pool is the sole parse path on a cache miss.
*/
readonly usedWorkerPool: boolean;
/** Files actually dispatched to parser workers after parse-cache lookup. */
readonly reparsedFileCount: number;
/**
* Per-file `ParsedFile` artifacts produced by workers' calls to
* `extractParsedFile`. Threaded through to `scopeResolutionPhase`

View file

@ -370,11 +370,13 @@ export const runPipelineFromRepo = async (
}
// Extract final results for the PipelineResult contract
const { totalFiles, usedWorkerPool, unavailableScopeLanguageFiles } = getPhaseOutput<{
totalFiles: number;
usedWorkerPool: boolean;
unavailableScopeLanguageFiles: number;
}>(results, 'parse');
const { totalFiles, usedWorkerPool, reparsedFileCount, unavailableScopeLanguageFiles } =
getPhaseOutput<{
totalFiles: number;
usedWorkerPool: boolean;
reparsedFileCount: number;
unavailableScopeLanguageFiles: number;
}>(results, 'parse');
let communityResult: CommunitiesOutput['communityResult'] | undefined;
let processResult: ProcessesOutput['processResult'] | undefined;
@ -426,6 +428,7 @@ export const runPipelineFromRepo = async (
resolutionOutcomes,
undecidedSatisfaction,
usedWorkerPool,
reparsedFileCount,
scopeExtractionFailures,
unavailableScopeLanguageFiles,
pdgEmitManifest,

View file

@ -13,6 +13,7 @@ import { detectGraphWriteCollapse, type GraphWriteCollapseVerdict } from './inde
import { PDG_EDGE_TYPES } from './lbug/pdg-emit-sink.js';
import path from 'path';
import fs from 'fs/promises';
import { constants as fsConstants } from 'node:fs';
import { randomUUID } from 'node:crypto';
import { retryRename } from '../storage/fs-atomic.js';
import { acquireIndexLock } from '../storage/index-lock.js';
@ -471,6 +472,29 @@ export interface AnalyzeOptions {
* Process exit reclaims the handles. Long-lived callers (MCP server, tests)
* leave this unset so they get a real close. See `closeLbug`. */
skipNativeCloseOnExit?: boolean;
/**
* Stage an incremental write in a copy of the live index before publishing
* it. Used by long-lived watch mode so a failed refresh leaves the previous
* graph readable. Currently supported on POSIX, where an open DB can be
* atomically renamed; Windows retains the established in-place path.
*/
atomicIncremental?: boolean;
}
const liveIndexMutationRisks = new WeakSet<object>();
function recordLiveIndexMutationRisk(error: unknown): void {
if ((typeof error === 'object' && error !== null) || typeof error === 'function') {
liveIndexMutationRisks.add(error);
}
}
/** Whether a failed analyze may already have changed the live DB. */
export function analyzeFailureMayHaveMutatedLiveIndex(error: unknown): boolean {
return (
((typeof error === 'object' && error !== null) || typeof error === 'function') &&
liveIndexMutationRisks.has(error)
);
}
export interface AnalyzeResult {
@ -522,6 +546,14 @@ export interface AnalyzeResult {
* (The historical "primary" name is kept it is public API surface.)
*/
isPrimaryBranch?: boolean;
/** Measured work performed by a successful incremental refresh. */
incrementalStats?: {
changedFiles: number;
reparsedFiles: number;
affectedDependents: number;
deletedFiles: number;
writeMode: 'incremental' | 'full';
};
}
/**
@ -1944,12 +1976,15 @@ async function runFullAnalysisInner(
process.platform === 'win32' &&
options.pdg !== true &&
process.env.GITNEXUS_ATOMIC_WINDOWS_SWAP === '1';
// Incremental atomicity copies the whole index into the temp before mutating
// it, which negates incremental's speed premise — so it is opt-in
// (GITNEXUS_ATOMIC_INCREMENTAL=1) pending a benchmark. Full rebuilds always
// swap where the platform allows.
// Incremental atomicity stages the whole index before mutation. It remains
// opt-in for ordinary analyze runs; watch mode requests it for failure
// preservation. The copy requests a filesystem clone and records its actual
// duration, while Node falls back to a normal copy where reflinks are absent.
const wantAtomicIncremental =
isIncremental && !!hashDiff && process.env.GITNEXUS_ATOMIC_INCREMENTAL === '1';
isIncremental &&
!!hashDiff &&
process.platform !== 'win32' &&
(options.atomicIncremental === true || process.env.GITNEXUS_ATOMIC_INCREMENTAL === '1');
// #2614 F3: the copy-then-swap stages ONLY the main lbug file, so a live index
// carrying an orphan .wal/.shadow (a silently-failed prior checkpoint) would
// be copied incompletely and lose that delta. Only take the atomic path when
@ -1967,6 +2002,10 @@ async function runFullAnalysisInner(
// valve. Nothing between here and there reads either binding except
// `initLbug(buildPath)`, which the upgrade re-runs against the staging path.
let useAtomicSwap = (isFullRebuild || atomicIncremental) && (posixSwap || windowsSwapOk);
// Set only at the first operation that can mutate the live graph store.
// Pre-write failures (config, lock, parsing, metadata, importer expansion)
// remain retryable even when this platform cannot use an atomic swap.
let liveIndexMutationStarted = false;
// #2658: a per-run staging name (was the fixed `lbug.new`). Even under the
// single-writer lock, a unique name means a crashed run's half-built staging
// file can never be mistaken for — or clobber — a live run's; the lock's
@ -1999,10 +2038,15 @@ async function runFullAnalysisInner(
if (atomicIncremental) {
// Stage the live index into the temp so the in-place delete/writeback
// below mutates the COPY, and the end-of-run swap publishes it atomically.
// Clear any stale temp first (a crashed run), then copy the (consolidated,
// single-file) live index. Whole-file copy — hence opt-in.
// Clear any stale temp first (a crashed run), then clone/copy the
// consolidated single-file live index.
await wipeLbugDbFiles(buildPath);
await fs.copyFile(lbugPath, buildPath);
const copyStartedAt = Date.now();
await fs.copyFile(lbugPath, buildPath, fsConstants.COPYFILE_FICLONE);
log(
`atomic-incremental: staged ${lbugPath} in ${Date.now() - copyStartedAt}ms ` +
'(copy-on-write requested; filesystem fallback is allowed)',
);
}
} else {
// Full rebuild path: wipe DB files first.
@ -2038,7 +2082,13 @@ async function runFullAnalysisInner(
// (`buildPath` = `<lbugPath>.new`, clearing any stragglers from a crashed
// run) and leaves the live index untouched until the end-of-run swap. On
// Windows buildPath === lbugPath, so this is the original in-place wipe.
await wipeLbugDbFiles(buildPath);
if (buildPath === lbugPath) liveIndexMutationStarted = true;
try {
await wipeLbugDbFiles(buildPath);
} catch (error) {
if (liveIndexMutationStarted) recordLiveIndexMutationRisk(error);
throw error;
}
}
// Size the buffer pool to the graph just built by the pipeline (a page cache
@ -2061,7 +2111,12 @@ async function runFullAnalysisInner(
// Full rebuild (POSIX) builds into the temp `buildPath`; incremental and
// Windows use `buildPath === lbugPath` in place.
await initLbug(buildPath);
try {
await initLbug(buildPath);
} catch (error) {
if (liveIndexMutationStarted) recordLiveIndexMutationRisk(error);
throw error;
}
// Manual WAL checkpoint driver (#1741): periodically drain the WAL
// from JS so the un-retriable native auto-checkpoint almost never
@ -2086,6 +2141,7 @@ async function runFullAnalysisInner(
// "escalated full write" (DB wiped, index destroyed) — tri-review
// 4669518496 P1.
let escalatedFullWrite = false;
let incrementalStats: AnalyzeResult['incrementalStats'];
// Phase 3.5's restore scope (FIX 3 of this shipping review): on the
// SURGICAL write plan this is the exact file set whose rows
// deleteNodesForFiles just removed — only THOSE files' cached embedding
@ -2211,6 +2267,13 @@ async function runFullAnalysisInner(
}
}
const importerExpansion = writableFiles.size - directlyChangedCount;
incrementalStats = {
changedFiles: hashDiff.changed.length + hashDiff.added.length + hashDiff.deleted.length,
reparsedFiles: pipelineResult.reparsedFileCount,
affectedDependents: importerExpansion,
deletedFiles: hashDiff.deleted.length,
writeMode: 'incremental',
};
await saveIncrementalDirtyState('importer-bfs', {
importerExpansion,
shadowSeedCount: shadowSeed.length,
@ -2572,6 +2635,7 @@ async function runFullAnalysisInner(
}
await walCheckpointDriver.stop();
await closeLbug();
if (buildPath === lbugPath) liveIndexMutationStarted = true;
await wipeLbugDbFiles(buildPath);
await initLbug(buildPath);
walCheckpointDriver = startWalCheckpointDriver();
@ -2597,6 +2661,7 @@ async function runFullAnalysisInner(
// same connection, and nothing on this branch creates or drops an index
// in between — so re-reading would only weaken the one-read invariant
// the snapshot type exists to enforce.
if (buildPath === lbugPath) liveIndexMutationStarted = true;
await dropSearchFTSIndexes(indexCatalogRows);
// 1b. Remove the write set's existing rows — batched (#2409): one
// DETACH DELETE per table per 200-file chunk. The former per-file
@ -3773,6 +3838,7 @@ async function runFullAnalysisInner(
: false;
if (useAtomicSwap && builtDbExists) {
await retryRename(buildPath, lbugPath);
liveIndexMutationStarted = true;
// Clear any sidecars orphaned beside the replaced file. A cleanly-closed
// prior index has none; a crashed one could, and it would be replay
// poison next to the freshly published index. Best-effort.
@ -3807,6 +3873,12 @@ async function runFullAnalysisInner(
ftsSkipped: !ftsReady,
ftsSkipReason: ftsReady ? undefined : ftsSkipReason,
isPrimaryBranch: !placement.branch,
incrementalStats: incrementalStats
? {
...incrementalStats,
writeMode: escalatedFullWrite ? 'full' : 'incremental',
}
: undefined,
};
} catch (err) {
// Ensure LadybugDB is closed even on error. Stop the driver first
@ -3845,6 +3917,11 @@ async function runFullAnalysisInner(
/* swallow — orphan reclamation must never mask the real failure */
}
}
if (liveIndexMutationStarted) {
// Preserve the original error identity/prototype: callers distinguish
// IndexLockTimeoutError and other domain failures with `instanceof`.
recordLiveIndexMutationRisk(err);
}
throw err;
}
}

View file

@ -917,7 +917,17 @@ export const persistParseCacheChunk = async (
createdCacheDirs.add(cacheDir);
}
const payload = JSON.stringify(slim, mapReplacer);
await fs.writeFile(getCacheChunkPath(cache.storagePath, chunkHash), payload, 'utf-8');
const chunkPath = getCacheChunkPath(cache.storagePath, chunkHash);
try {
await fs.writeFile(chunkPath, payload, 'utf-8');
} catch (error) {
if ((error as NodeJS.ErrnoException).code !== 'ENOENT') throw error;
// Long-lived analyze --watch processes can replace the sharded cache
// directory after this process-local memo recorded it as created.
await fs.mkdir(cacheDir, { recursive: true });
createdCacheDirs.add(cacheDir);
await fs.writeFile(chunkPath, payload, 'utf-8');
}
cache.onDiskKeys ??= new Set<string>();
cache.onDiskKeys.add(chunkHash);
cache.entries.delete(chunkHash);

View file

@ -40,6 +40,8 @@ export interface PipelineResult {
* affordance so regression suites can prove the pool engaged.
*/
usedWorkerPool: boolean;
/** Files actually dispatched to parser workers after parse-cache lookup. */
reparsedFileCount: number;
/** Files omitted from scope-resolution while the rest of analysis continued. */
scopeExtractionFailures: readonly string[];
/** Files scope resolution could not inspect because their parser was unavailable. */

View file

@ -22,17 +22,28 @@ type LbugAdapter = typeof import('../../src/core/lbug/lbug-adapter.js');
const ctx = vi.hoisted(() => ({
loadMock: vi.fn(),
realLoad: null as LbugAdapter['loadGraphToLbug'] | null,
deleteMock: vi.fn(),
realDelete: null as LbugAdapter['deleteNodesForFiles'] | null,
}));
// Delegating mock: overrides only loadGraphToLbug so a rebuild can be made to
// fail on demand (mirrors run-analyze-adopt-failure.test.ts).
vi.mock('../../src/core/lbug/lbug-adapter.js', async (importOriginal) => {
const actual = await importOriginal<LbugAdapter>();
ctx.realLoad = actual.loadGraphToLbug;
ctx.realDelete = actual.deleteNodesForFiles;
ctx.loadMock.mockImplementation(actual.loadGraphToLbug);
return { ...actual, loadGraphToLbug: ctx.loadMock };
ctx.deleteMock.mockImplementation(actual.deleteNodesForFiles);
return {
...actual,
loadGraphToLbug: ctx.loadMock,
deleteNodesForFiles: ctx.deleteMock,
};
});
import { runFullAnalysis } from '../../src/core/run-analyze.js';
import {
analyzeFailureMayHaveMutatedLiveIndex,
runFullAnalysis,
} from '../../src/core/run-analyze.js';
import { getStoragePaths } from '../../src/storage/repo-manager.js';
import {
initLbug as poolInit,
@ -68,6 +79,10 @@ describe.skipIf(isWin)('atomic full-rebuild swap (#2)', () => {
ctx.loadMock.mockImplementation((...a: Parameters<LbugAdapter['loadGraphToLbug']>) =>
ctx.realLoad!(...a),
);
ctx.deleteMock.mockReset();
ctx.deleteMock.mockImplementation((...a: Parameters<LbugAdapter['deleteNodesForFiles']>) =>
ctx.realDelete!(...a),
);
});
afterEach(async () => {
@ -129,6 +144,29 @@ describe.skipIf(isWin)('atomic full-rebuild swap (#2)', () => {
}
}, 180_000);
it('marks a failure after an atomic publish as potentially live-mutating', async () => {
const { repo, cleanup } = await makeRepo();
try {
const failure = await runFullAnalysis(
repo,
{},
{
onProgress: (phase, percent) => {
if (phase === 'done' && percent === 100) {
throw new Error('injected post-publish failure');
}
},
},
).catch((error: unknown) => error);
expect(failure).toMatchObject({ message: 'injected post-publish failure' });
expect(analyzeFailureMayHaveMutatedLiveIndex(failure)).toBe(true);
await expect(fs.stat(getStoragePaths(repo).lbugPath)).resolves.toBeTruthy();
} finally {
await cleanup();
}
}, 180_000);
it('the read pool serves the freshly-swapped index after a rebuild (#1 + #2 end-to-end)', async () => {
const { repo, cleanup } = await makeRepo();
const repoId = 'atomic-swap-e2e';
@ -202,6 +240,76 @@ describe.skipIf(isWin)('atomic full-rebuild swap (#2)', () => {
}
}, 180_000);
it('keeps the live graph unchanged when atomic incremental writeback fails', async () => {
const { repo, cleanup } = await makeRepo();
const repoId = 'atomic-incr-failure';
try {
await runFullAnalysis(repo, {}, { onProgress: () => {} });
const { lbugPath } = getStoragePaths(repo);
const before = await identity(lbugPath);
await fs.writeFile(
path.join(repo, 'a.ts'),
'export function greet(n: string) { return `hi ${n}`; }\nexport function caller() { return greet("x"); }\nexport function addedAfterRetry() { return 1; }\n',
);
execSync('git -c user.name=t -c user.email=t@t commit -am change', {
cwd: repo,
stdio: 'pipe',
});
ctx.deleteMock.mockRejectedValueOnce(new Error('injected incremental write failure'));
const failure = await runFullAnalysis(
repo,
{ atomicIncremental: true },
{ onProgress: () => {} },
).catch((error: unknown) => error);
expect(failure).toMatchObject({ message: 'injected incremental write failure' });
expect(analyzeFailureMayHaveMutatedLiveIndex(failure)).toBe(false);
expect(await identity(lbugPath)).toBe(before);
expect(await lingeringTemp(lbugPath)).toEqual([]);
await poolInit(repoId, lbugPath);
const beforeRetry = (
await poolQuery(repoId, 'MATCH (f:Function) RETURN f.name AS n')
).flatMap((row) => Object.values(row as Record<string, unknown>).map(String));
expect(beforeRetry).toContain('greet');
expect(beforeRetry).not.toContain('addedAfterRetry');
await poolClose(repoId);
await runFullAnalysis(repo, { atomicIncremental: true }, { onProgress: () => {} });
await poolInit(repoId, lbugPath);
const afterRetry = (await poolQuery(repoId, 'MATCH (f:Function) RETURN f.name AS n')).flatMap(
(row) => Object.values(row as Record<string, unknown>).map(String),
);
expect(afterRetry).toContain('addedAfterRetry');
} finally {
await poolClose(repoId);
await cleanup();
}
}, 180_000);
it('marks failed in-place incremental writes as potentially live-mutating', async () => {
const { repo, cleanup } = await makeRepo();
try {
await runFullAnalysis(repo, {}, { onProgress: () => {} });
await fs.writeFile(path.join(repo, 'a.ts'), 'export function changed() { return 1; }\n');
execSync('git -c user.name=t -c user.email=t@t commit -am change', {
cwd: repo,
stdio: 'pipe',
});
ctx.deleteMock.mockRejectedValueOnce(new Error('injected in-place failure'));
const failure = await runFullAnalysis(repo, {}, { onProgress: () => {} }).catch(
(error: unknown) => error,
);
expect(failure).toBeInstanceOf(Error);
expect(failure).toMatchObject({ message: 'injected in-place failure' });
expect(analyzeFailureMayHaveMutatedLiveIndex(failure)).toBe(true);
} finally {
await cleanup();
}
}, 180_000);
it('publishes cleanly on the production close path (skipNativeCloseOnExit) (#2614 F5)', async () => {
const { repo, cleanup } = await makeRepo();
try {

View file

@ -1048,6 +1048,198 @@ describe('CLI end-to-end', () => {
expect(result.stdout).toMatch(/analyze|status|serve/i);
});
it('shows the analyze watch mode and its debounce controls', () => {
const result = runCliRaw(['analyze', '--help'], MINI_REPO);
expect(result.status).toBe(0);
expect(result.stdout).toContain('--watch');
expect(result.stdout).toContain('--debounce');
expect(result.stdout).toContain('--workers');
});
it('rejects --debounce without --watch', () => {
const result = runCliRaw(['analyze', '--debounce', '25'], MINI_REPO);
expect(result.status).toBe(1);
expect(result.stderr).toContain('--debounce requires --watch');
});
it('runs production analyze --watch with exact telemetry and transactional config reloads', async () => {
const repo = makeMiniRepoCopy('watch-repo', 'gn-watch-cli-');
const home = fs.mkdtempSync(path.join(os.tmpdir(), 'gn-watch-cli-home-'));
try {
fs.writeFileSync(
path.join(repo, '.gitnexusrc'),
JSON.stringify({ workers: '1', maxFileSize: '1' }),
'utf8',
);
await new Promise<void>((resolve, reject) => {
const child = spawn(
process.execPath,
[...CLI_SPAWN_PREFIX, 'analyze', repo, '--watch', '--debounce', '25', '--workers', '1'],
{
cwd: repo,
stdio: ['ignore', 'pipe', 'pipe'],
env: cliEnv({ GITNEXUS_HOME: home }),
},
);
let stdout = '';
let stderr = '';
let transcript = '';
let baselineNodes: number | undefined;
let stage = 'ready';
let stageOffset = 0;
let settled = false;
const timer = setTimeout(() => {
if (settled) return;
settled = true;
child.kill('SIGTERM');
reject(new Error(`watch CLI timed out\nstdout:\n${stdout}\nstderr:\n${stderr}`));
}, 480_000);
const advance = (nextStage: string, action: () => void) => {
stage = nextStage;
stageOffset = transcript.length;
setTimeout(action, 200);
};
const writeLargeSource = (fileName: string, functionName: string) => {
fs.writeFileSync(
path.join(repo, fileName),
`const padding = '${'x'.repeat(1_500)}';\n` +
`export function ${functionName}(): number { return padding.length; }\n`,
'utf8',
);
};
const handleOutput = () => {
const output = transcript.slice(stageOffset);
if (stage === 'ready' && /Watching .*index (?:is up to date|ready)/.test(output)) {
const meta = JSON.parse(
fs.readFileSync(path.join(repo, '.gitnexus', 'gitnexus.json'), 'utf8'),
);
baselineNodes = meta.stats.nodes;
advance('proof', () => {
fs.writeFileSync(
path.join(repo, 'watch-proof.ts'),
'export function watchProof(): number { return 1; }\n',
'utf8',
);
});
return;
}
if (stage === 'proof' && /Refresh complete: 1 changed, 8 re-parsed,/.test(output)) {
const meta = JSON.parse(
fs.readFileSync(path.join(repo, '.gitnexus', 'gitnexus.json'), 'utf8'),
);
expect(meta.stats.nodes).toBeGreaterThan(baselineNodes!);
advance('first-large-file', () =>
writeLargeSource('oversized-before.ts', 'skippedByLimit'),
);
return;
}
if (
stage === 'first-large-file' &&
output.includes('Skipped 1 large files (>1KB)') &&
output.includes('- oversized-before.ts') &&
/Refresh complete: 0 changed,/.test(output)
) {
advance('invalid-config', () => {
fs.writeFileSync(
path.join(repo, '.gitnexusrc'),
JSON.stringify({ workers: '1', maxFileSize: '0' }),
'utf8',
);
});
return;
}
if (
stage === 'invalid-config' &&
/Refresh failed.*maxFileSize must be a positive integer/.test(output)
) {
advance('second-large-file', () =>
writeLargeSource('oversized-after-invalid.ts', 'stillSkipped'),
);
return;
}
if (
stage === 'second-large-file' &&
/Refresh failed.*Configuration remains invalid/.test(output)
) {
advance('recovered-config', () => {
fs.writeFileSync(
path.join(repo, '.gitnexusrc'),
JSON.stringify({ workers: '1', maxFileSize: '4096' }),
'utf8',
);
});
return;
}
if (
stage === 'recovered-config' &&
/Refresh complete: [2-9][0-9]* changed, [1-9][0-9]* re-parsed,/.test(output)
) {
stage = 'stopping';
setTimeout(() => child.kill('SIGTERM'), 100);
}
};
child.stderr.on('data', (chunk: Buffer) => {
const text = chunk.toString();
stderr += text;
transcript += text;
handleOutput();
});
child.stdout.on('data', (chunk: Buffer) => {
const text = chunk.toString();
stdout += text;
transcript += text;
handleOutput();
});
child.once('error', (error) => {
if (settled) return;
settled = true;
clearTimeout(timer);
reject(error);
});
child.once('close', (code, signal) => {
if (settled) return;
settled = true;
clearTimeout(timer);
const expectedWindowsTermination =
process.platform === 'win32' && code === null && signal === 'SIGTERM';
if (code !== 0 && !expectedWindowsTermination) {
reject(
new Error(
`watch CLI exited ${code ?? signal}\nstdout:\n${stdout}\nstderr:\n${stderr}`,
),
);
return;
}
expect(stage).toBe('stopping');
expect(transcript).toContain('Refresh complete: 1 changed, 8 re-parsed,');
resolve();
});
});
for (const [symbol, file] of [
['watchProof', 'watch-proof.ts'],
['skippedByLimit', 'oversized-before.ts'],
['stillSkipped', 'oversized-after-invalid.ts'],
]) {
const result = runCliWithEnv(
['context', symbol, '--file', file],
repo,
{ GITNEXUS_HOME: home },
30_000,
);
expect(result.status).toBe(0);
expect(result.stdout).toContain(symbol);
expect(result.stdout).toContain(file);
}
} finally {
cleanupTempDirSync(path.dirname(repo));
cleanupTempDirSync(home);
}
}, 540_000);
it('fails with unknown command', () => {
const result = runCliRaw(['nonexistent'], MINI_REPO);

View file

@ -0,0 +1,230 @@
import { execFileSync } from 'node:child_process';
import fs from 'node:fs/promises';
import os from 'node:os';
import path from 'node:path';
import { afterEach, describe, expect, it, vi } from 'vitest';
import { startWatchFileLoop, type WatchFileLoop } from '../../src/cli/watch.js';
import { cleanupTempDir } from '../helpers/test-db.js';
const tempDirs: string[] = [];
const loops: WatchFileLoop[] = [];
async function waitFor(predicate: () => boolean, timeoutMs = 5_000): Promise<void> {
const deadline = Date.now() + timeoutMs;
while (!predicate()) {
if (Date.now() >= deadline) throw new Error('timed out waiting for watcher event');
await new Promise((resolve) => setTimeout(resolve, 25));
}
}
async function makeRepo(): Promise<string> {
const repo = await fs.mkdtemp(path.join(os.tmpdir(), 'gitnexus-watch-fs-'));
tempDirs.push(repo);
execFileSync('git', ['init', '-q'], { cwd: repo });
return repo;
}
afterEach(async () => {
await Promise.all(loops.splice(0).map((loop) => loop.close()));
await Promise.all(tempDirs.splice(0).map((dir) => cleanupTempDir(dir)));
});
describe('watch filesystem integration', () => {
it('fails startup and closes the watcher when the initial analysis fails', async () => {
const repo = await makeRepo();
const onError = vi.fn();
await expect(
startWatchFileLoop(
repo,
25,
async () => {
throw new Error('initial analysis failed');
},
onError,
),
).rejects.toThrow('initial analysis failed');
expect(onError).not.toHaveBeenCalled();
});
it('never enqueues analyzer-owned .gitnexus writes created by the initial refresh', async () => {
const repo = await makeRepo();
const batches: string[][] = [];
const loop = await startWatchFileLoop(
repo,
25,
async (paths) => {
batches.push([...paths]);
if (paths.length === 0) {
await fs.mkdir(path.join(repo, '.gitnexus'), { recursive: true });
await fs.writeFile(path.join(repo, '.gitnexus', 'gitnexus.json'), '{}\n', 'utf8');
await fs.writeFile(path.join(repo, '.gitnexus', 'lbug'), 'index bytes', 'utf8');
}
},
(error) => {
throw error;
},
);
loops.push(loop);
await new Promise((resolve) => setTimeout(resolve, 200));
await loop.waitForIdle();
expect(batches).toEqual([[]]);
});
it('coalesces indexed add/change/rename/delete events and stops cleanly', async () => {
const repo = await makeRepo();
const batches: string[][] = [];
const loop = await startWatchFileLoop(
repo,
30,
async (paths) => batches.push([...paths]),
(error) => {
throw error;
},
);
loops.push(loop);
expect(batches).toEqual([[]]);
await fs.writeFile(path.join(repo, 'README.md'), '# One', 'utf8');
await fs.writeFile(path.join(repo, 'src.ts'), 'export const one = 1;', 'utf8');
await fs.writeFile(path.join(repo, 'src.ts'), 'export const one = 2;', 'utf8');
await waitFor(() => batches.flat().includes('README.md') && batches.flat().includes('src.ts'));
await fs.rename(path.join(repo, 'src.ts'), path.join(repo, 'renamed.ts'));
await waitFor(() => batches.flat().includes('renamed.ts'));
await fs.rm(path.join(repo, 'renamed.ts'));
await waitFor(() => batches.flat().filter((entry) => entry === 'renamed.ts').length >= 2);
expect(batches.flat()).toEqual(expect.arrayContaining(['README.md', 'src.ts', 'renamed.ts']));
await loop.close();
loops.pop();
const countAfterClose = batches.length;
await fs.writeFile(path.join(repo, 'after-close.ts'), 'export {};', 'utf8');
await new Promise((resolve) => setTimeout(resolve, 150));
expect(batches).toHaveLength(countAfterClose);
});
it('queues edits during refresh, recovers after failure, and ignores external symlinks', async () => {
const repo = await makeRepo();
const outside = await fs.mkdtemp(path.join(os.tmpdir(), 'gitnexus-watch-outside-'));
tempDirs.push(outside);
await fs.symlink(
outside,
path.join(repo, 'external'),
process.platform === 'win32' ? 'junction' : 'dir',
);
const successful: string[][] = [];
const errors: string[][] = [];
let failNext = false;
let releaseRefresh: (() => void) | undefined;
const loop = await startWatchFileLoop(
repo,
25,
async (paths) => {
if (failNext) {
failNext = false;
throw new Error('injected refresh failure');
}
successful.push([...paths]);
if (paths.includes('first.ts')) {
await new Promise<void>((resolve) => {
releaseRefresh = resolve;
});
}
},
(_error, paths) => errors.push([...paths]),
);
loops.push(loop);
await fs.writeFile(path.join(repo, 'first.ts'), 'export const first = 1;', 'utf8');
await waitFor(() => releaseRefresh !== undefined);
await fs.writeFile(path.join(repo, 'during.ts'), 'export const during = 1;', 'utf8');
releaseRefresh!();
await waitFor(() => successful.flat().includes('during.ts'));
failNext = true;
await fs.writeFile(path.join(repo, 'fails.ts'), 'export const fail = 1;', 'utf8');
await waitFor(() => errors.length === 1);
await waitFor(() => successful.flat().includes('fails.ts'));
await fs.writeFile(path.join(repo, 'retry.ts'), 'export const retry = 1;', 'utf8');
await waitFor(() => successful.flat().includes('retry.ts'));
const beforeExternal = successful.length + errors.length;
await fs.writeFile(path.join(outside, 'outside.ts'), 'export const outside = 1;', 'utf8');
await new Promise((resolve) => setTimeout(resolve, 200));
expect(successful.length + errors.length).toBe(beforeExternal);
});
it('reloads gitignore rules before processing subsequent file events', async () => {
const repo = await makeRepo();
await fs.writeFile(path.join(repo, '.gitignore'), 'blocked.ts\n', 'utf8');
const batches: string[][] = [];
const loop = await startWatchFileLoop(
repo,
25,
async (paths) => batches.push([...paths]),
(error) => {
throw error;
},
);
loops.push(loop);
await fs.writeFile(path.join(repo, 'blocked.ts'), 'export const blocked = 1;', 'utf8');
await new Promise((resolve) => setTimeout(resolve, 200));
expect(batches.flat()).not.toContain('blocked.ts');
await fs.writeFile(path.join(repo, '.gitignore'), '', 'utf8');
await waitFor(() => batches.flat().includes('.gitignore'));
await fs.writeFile(path.join(repo, 'blocked.ts'), 'export const blocked = 2;', 'utf8');
await waitFor(() => batches.flat().includes('blocked.ts'));
});
it('keeps the last valid ignore predicate after an oversized reload and later recovers', async () => {
const repo = await makeRepo();
await fs.writeFile(path.join(repo, '.gitignore'), 'blocked.ts\n', 'utf8');
const batches: string[][] = [];
const errors: string[][] = [];
const loop = await startWatchFileLoop(
repo,
25,
async (paths) => batches.push([...paths]),
(_error, paths) => errors.push([...paths]),
);
loops.push(loop);
await fs.writeFile(path.join(repo, '.gitignore'), 'x'.repeat(1024 * 1024 + 1), 'utf8');
await waitFor(() => errors.flat().includes('.gitignore'));
await waitFor(() => errors.length >= 2);
await fs.writeFile(path.join(repo, 'other.ts'), 'export const other = 1;', 'utf8');
await waitFor(() => errors.flat().includes('other.ts'));
expect(batches.flat()).not.toContain('other.ts');
await fs.writeFile(path.join(repo, 'blocked.ts'), 'export const blocked = 1;', 'utf8');
await new Promise((resolve) => setTimeout(resolve, 200));
expect(batches.flat()).not.toContain('blocked.ts');
await fs.writeFile(path.join(repo, '.gitignore'), '', 'utf8');
await waitFor(() => batches.flat().includes('.gitignore'));
await fs.writeFile(path.join(repo, 'blocked.ts'), 'export const blocked = 2;', 'utf8');
await waitFor(() => batches.flat().includes('blocked.ts'));
});
it('observes root control files even when gitignore excludes them', async () => {
const repo = await makeRepo();
await fs.writeFile(path.join(repo, '.gitignore'), '.gitnexusrc\n', 'utf8');
const batches: string[][] = [];
const loop = await startWatchFileLoop(
repo,
25,
async (paths) => batches.push([...paths]),
(error) => {
throw error;
},
);
loops.push(loop);
await fs.writeFile(path.join(repo, '.gitnexusrc'), '{}\n', 'utf8');
await waitFor(() => batches.flat().includes('.gitnexusrc'));
});
});

View file

@ -1,9 +1,12 @@
import { describe, it, expect, beforeEach, afterEach } from 'vitest';
import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest';
import { execFileSync } from 'node:child_process';
import fsSync from 'node:fs';
import fs from 'fs/promises';
import path from 'path';
import os from 'os';
import {
loadAnalyzeConfig,
loadAnalyzeConfigStrict,
mergeAnalyzeOptions,
resolveDefaultBranch,
validateBranchName,
@ -13,6 +16,10 @@ import {
DEFAULT_BRANCH_FALLBACK,
} from '../../src/cli/analyze-config.js';
import type { AnalyzeOptions } from '../../src/cli/analyze.js';
import {
MAX_REPO_CONTROL_FILE_BYTES,
readRepoControlFile,
} from '../../src/config/repo-control-file.js';
describe('analyze-config (.gitnexusrc support, #243)', () => {
let dir: string;
@ -34,6 +41,77 @@ describe('analyze-config (.gitnexusrc support, #243)', () => {
expect(loadAnalyzeConfig(dir)).toBeUndefined();
});
it('rejects an oversized repository config before parsing', async () => {
await writeRc(' '.repeat(MAX_REPO_CONTROL_FILE_BYTES + 1));
await expect(loadAnalyzeConfigStrict(dir)).rejects.toThrow(/exceeds/);
});
it('keeps the read bounded if a control file grows after its size check', async () => {
await writeRc('{}');
const fstatSync = fsSync.fstatSync;
const stat = vi.spyOn(fsSync, 'fstatSync').mockImplementation((fd) => {
const opened = fstatSync(fd);
Object.defineProperty(opened, 'size', { value: MAX_REPO_CONTROL_FILE_BYTES + 1 });
return opened;
});
try {
await expect(readRepoControlFile(dir, GITNEXUS_RC_FILENAME)).rejects.toThrow(/exceeds/);
expect(stat).toHaveBeenCalledOnce();
} finally {
stat.mockRestore();
}
});
it('rejects a hardlinked repository config', async () => {
const outside = await fs.mkdtemp(path.join(os.tmpdir(), 'gn-rc-hardlink-'));
try {
const target = path.join(outside, 'config.json');
await fs.writeFile(target, JSON.stringify({ workers: '8' }));
await fs.link(target, path.join(dir, GITNEXUS_RC_FILENAME));
await expect(loadAnalyzeConfigStrict(dir)).rejects.toThrow(/hard link/);
} finally {
await fs.rm(outside, { recursive: true, force: true });
}
});
it.skipIf(process.platform === 'win32')(
'rejects a FIFO before opening it for reading',
async () => {
const fifo = path.join(dir, GITNEXUS_RC_FILENAME);
execFileSync('mkfifo', [fifo]);
let timeout: ReturnType<typeof setTimeout> | undefined;
try {
await expect(
Promise.race([
readRepoControlFile(dir, GITNEXUS_RC_FILENAME),
new Promise<never>((_resolve, reject) => {
timeout = setTimeout(() => reject(new Error('FIFO read did not fail promptly')), 500);
}),
]),
).rejects.toThrow(/regular file/);
} finally {
if (timeout !== undefined) clearTimeout(timeout);
}
},
);
it.skipIf(process.platform === 'win32')(
'rejects a final-file symlink for repository config',
async () => {
const outside = await fs.mkdtemp(path.join(os.tmpdir(), 'gn-rc-outside-'));
try {
const target = path.join(outside, 'config.json');
await fs.writeFile(target, JSON.stringify({ workers: '8' }));
await fs.symlink(target, path.join(dir, GITNEXUS_RC_FILENAME), 'file');
await expect(loadAnalyzeConfigStrict(dir)).rejects.toThrow(/symbolic link/);
} finally {
await fs.rm(outside, { recursive: true, force: true });
}
},
);
it('throws an actionable error on invalid JSON, naming the file', async () => {
await writeRc('{ not valid json ');
expect(() => loadAnalyzeConfig(dir)).toThrow(GitNexusRcError);

View file

@ -220,6 +220,15 @@ describe('analyzeCommand heap respawn', () => {
expect(parseMaxOldSpaceMb('--max-old-space-size --other-flag')).toBeNull();
});
it('preserves conventional signal exits for analyze but treats watch shutdown as clean', async () => {
const { forwardedSignalExitCode } = await import('../../src/cli/analyze.js');
expect(forwardedSignalExitCode('SIGINT', false)).toBe(130);
expect(forwardedSignalExitCode('SIGTERM', false)).toBe(143);
expect(forwardedSignalExitCode('SIGINT', true)).toBe(0);
expect(forwardedSignalExitCode('SIGTERM', true)).toBe(0);
expect(forwardedSignalExitCode('SIGABRT', false)).toBe(1);
});
it('GITNEXUS_MEMORY=off also disables the default (unpinned) respawn (#2649 review)', async () => {
delete process.env.NODE_OPTIONS;
process.env.GITNEXUS_MEMORY = 'off';

View file

@ -3,7 +3,7 @@
*
* The regression this guards is specific and was expensive: three CHEAP files
* were registered in `SPAWN_CLI`, vitest re-partitioned the list by file COUNT,
* and the reshuffle clustered `cli-e2e` (361 s on Windows) with `cli-limit-e2e`
* and the reshuffle clustered `cli-e2e` (now 621 s on Windows) with `cli-limit-e2e`
* (75 s) and `analyze-heap-oom-e2e` (23 s) on one shard, which then blew the
* 20-minute watchdog. The added files cost nothing; the COUNT-split did it.
*
@ -49,7 +49,7 @@ describe('cross-platform shard partition', () => {
});
it('never puts the two heaviest suites on the same shard', () => {
// The exact shape of the outage: cli-e2e and worker-pool are 361 s and
// The exact shape of the outage: cli-e2e and worker-pool are 621 s and
// 222 s, so together they are most of a shard's budget before anything else
// is scheduled.
const shards = allShards(ALL_CROSS_PLATFORM, SHARD_TOTAL);

View file

@ -842,6 +842,13 @@ describe('runFullAnalysis — incremental orchestration', () => {
{ onProgress: () => {} },
);
expect(incremental.alreadyUpToDate).toBeUndefined();
expect(incremental.incrementalStats).toMatchObject({
changedFiles: 1,
affectedDependents: 2,
deletedFiles: 0,
writeMode: 'incremental',
});
expect(incremental.incrementalStats?.reparsedFiles).toBe(7);
expect(
querySpy.mock.calls.some(
([query]) =>

View file

@ -748,4 +748,31 @@ describe('loadParseCache / saveParseCache (round-trip)', () => {
await rm(dir, { recursive: true, force: true });
}
});
it('recreates a memoized shard directory after a long-lived process replaces it', async () => {
const dir = await mkdtemp(path.join(tmpdir(), 'gnx-pc-'));
try {
const firstKey = 'd'.repeat(64);
const secondKey = 'e'.repeat(64);
const cache: ParseCache = {
version: PARSE_CACHE_VERSION,
entries: new Map(),
usedKeys: new Set([firstKey]),
storagePath: dir,
onDiskKeys: new Set(),
};
await persistParseCacheChunk(cache, firstKey, [minimalResult({ fileCount: 1 })]);
await rm(path.join(dir, 'parse-cache'), { recursive: true, force: true });
cache.usedKeys = new Set([secondKey]);
await persistParseCacheChunk(cache, secondKey, [minimalResult({ fileCount: 2 })]);
await saveParseCache(dir, cache);
const loaded = await loadParseCache(dir);
expect((await loadParseCacheChunk(loaded, secondKey))?.[0]?.fileCount).toBe(2);
} finally {
await rm(dir, { recursive: true, force: true });
}
});
});

View file

@ -0,0 +1,29 @@
import { beforeEach, describe, expect, it, vi } from 'vitest';
const analyzeFailureMayHaveMutatedLiveIndex = vi.hoisted(() => vi.fn());
vi.mock('../../src/core/run-analyze.js', () => ({
analyzeFailureMayHaveMutatedLiveIndex,
runFullAnalysis: vi.fn(),
}));
import { shouldStopAfterWatchRefreshFailure } from '../../src/cli/watch.js';
describe('watch refresh failure policy', () => {
beforeEach(() => analyzeFailureMayHaveMutatedLiveIndex.mockReset());
it('retries a queued pre-write failure even when incremental writes are in-place', () => {
const error = new Error('failed before live graph mutation');
analyzeFailureMayHaveMutatedLiveIndex.mockReturnValue(false);
expect(shouldStopAfterWatchRefreshFailure(error, ['src/a.ts'])).toBe(false);
});
it('stops only when a queued failure may have mutated the live graph', () => {
const error = new Error('failed during live graph mutation');
analyzeFailureMayHaveMutatedLiveIndex.mockReturnValue(true);
expect(shouldStopAfterWatchRefreshFailure(error, ['src/a.ts'])).toBe(true);
expect(shouldStopAfterWatchRefreshFailure(error, [])).toBe(false);
});
});

View file

@ -0,0 +1,180 @@
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import fs from 'node:fs/promises';
import os from 'node:os';
import path from 'node:path';
import { createWatchIgnorePredicate } from '../../src/config/ignore-service.js';
import { isRelevantWatchPath, resolveWatchOptions } from '../../src/cli/watch.js';
import * as git from '../../src/storage/git.js';
vi.mock('../../src/storage/git.js', () => ({
getCoreExcludesFilePath: vi.fn(),
getGitInfoExcludePath: vi.fn(),
}));
let repoPath: string;
beforeEach(async () => {
repoPath = await fs.mkdtemp(path.join(os.tmpdir(), 'gitnexus-watch-'));
vi.mocked(git.getCoreExcludesFilePath).mockReturnValue(null);
vi.mocked(git.getGitInfoExcludePath).mockReturnValue(null);
});
afterEach(async () => {
await fs.rm(repoPath, { recursive: true, force: true });
});
describe('watch path selection', () => {
it('accepts every scanner-admitted file instead of maintaining a second allow-list', () => {
expect(isRelevantWatchPath('src/service.ts')).toBe(true);
expect(isRelevantWatchPath('server/app.py')).toBe(true);
expect(isRelevantWatchPath('backend/project.csproj')).toBe(true);
expect(isRelevantWatchPath('.gitnexusrc')).toBe(true);
expect(isRelevantWatchPath('README.md')).toBe(true);
expect(isRelevantWatchPath('docs/guide.mdx')).toBe(true);
expect(isRelevantWatchPath('config/application-prod.yml')).toBe(true);
expect(isRelevantWatchPath('src/main/resources/application.properties')).toBe(true);
expect(isRelevantWatchPath('templates/page.html')).toBe(true);
expect(isRelevantWatchPath('templates/page.htm')).toBe(true);
expect(isRelevantWatchPath('views/page.ejs')).toBe(true);
expect(isRelevantWatchPath('views/page.hbs')).toBe(true);
expect(isRelevantWatchPath('views/page.blade.php')).toBe(true);
expect(
isRelevantWatchPath(
'src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports',
),
).toBe(true);
expect(isRelevantWatchPath('src/main/resources/META-INF/spring.factories')).toBe(true);
expect(isRelevantWatchPath('tsconfig.base.json')).toBe(true);
expect(isRelevantWatchPath('packages/api/tsconfig.build.json')).toBe(true);
expect(isRelevantWatchPath('schema.sql')).toBe(true);
expect(isRelevantWatchPath('Dockerfile')).toBe(true);
expect(isRelevantWatchPath('assets/logo.png')).toBe(true);
expect(isRelevantWatchPath('../outside.ts')).toBe(false);
expect(isRelevantWatchPath('C:\\outside.ts')).toBe(false);
});
it('honors hardcoded, gitignore, and explicit-unignore rules', async () => {
await fs.writeFile(
path.join(repoPath, '.gitignore'),
['generated/*', '!generated/', '!generated/keep.ts'].join('\n'),
);
const ignored = await createWatchIgnorePredicate(repoPath);
expect(ignored(path.join(repoPath, 'node_modules', 'pkg', 'index.ts'))).toBe(true);
expect(ignored(path.join(repoPath, 'generated'), true)).toBe(false);
expect(ignored(path.join(repoPath, 'generated', 'drop.ts'))).toBe(true);
expect(ignored(path.join(repoPath, 'generated', 'keep.ts'))).toBe(false);
expect(ignored(path.join(repoPath, 'src', 'keep.ts'))).toBe(false);
expect(ignored(path.resolve(repoPath, '..', 'outside.ts'))).toBe(true);
});
it('does not partially mutate environment state when a reloaded config is invalid', async () => {
const names = [
'GITNEXUS_MAX_FILE_SIZE',
'GITNEXUS_WORKER_SUB_BATCH_TIMEOUT_MS',
'GITNEXUS_VERBOSE',
] as const;
const original = Object.fromEntries(names.map((name) => [name, process.env[name]]));
try {
await fs.writeFile(
path.join(repoPath, '.gitnexusrc'),
JSON.stringify({ maxFileSize: '2048', workerTimeout: '90', workers: '2' }),
);
const baseline = { maxFileSize: '512', workerTimeout: '30000', verbose: undefined };
await resolveWatchOptions(repoPath, {}, baseline);
expect(process.env.GITNEXUS_MAX_FILE_SIZE).toBe('2048');
expect(process.env.GITNEXUS_WORKER_SUB_BATCH_TIMEOUT_MS).toBe('90000');
await fs.writeFile(
path.join(repoPath, '.gitnexusrc'),
JSON.stringify({ maxFileSize: '4096', workerTimeout: '120', workers: '0' }),
);
await expect(resolveWatchOptions(repoPath, {}, baseline)).rejects.toThrow(
'--workers must be a positive integer',
);
expect(process.env.GITNEXUS_MAX_FILE_SIZE).toBe('2048');
expect(process.env.GITNEXUS_WORKER_SUB_BATCH_TIMEOUT_MS).toBe('90000');
} finally {
for (const name of names) {
const value = original[name];
if (value === undefined) delete process.env[name];
else process.env[name] = value;
}
}
});
it('ignores unsupported repository defaults but rejects explicit unsupported CLI flags', async () => {
await fs.writeFile(
path.join(repoPath, '.gitnexusrc'),
JSON.stringify({
embeddings: true,
defaultBranch: 'develop',
skipAgentsMd: false,
skipSkills: false,
stats: true,
}),
);
const ignored: string[][] = [];
await expect(
resolveWatchOptions(
repoPath,
{},
{
maxFileSize: undefined,
workerTimeout: undefined,
verbose: undefined,
},
(names) => ignored.push([...names]),
),
).resolves.toMatchObject({ skipAgentsMd: true, skipSkills: true });
expect(ignored).toEqual([
['embeddings', 'defaultBranch', 'skipAgentsMd', 'skipSkills', 'stats'],
]);
const unsupportedCliOptions: Array<[Parameters<typeof resolveWatchOptions>[1], string]> = [
[{ embeddings: true }, '--embeddings'],
[{ defaultBranch: 'develop' }, '--default-branch'],
[{ skipAgentsMd: true }, '--skip-agents-md'],
[{ skipSkills: true }, '--skip-skills'],
[{ stats: false }, '--no-stats'],
];
for (const [options, flag] of unsupportedCliOptions) {
await expect(
resolveWatchOptions(repoPath, options, {
maxFileSize: undefined,
workerTimeout: undefined,
verbose: undefined,
}),
).rejects.toThrow(`analyze --watch does not support ${flag}`);
}
});
it('rejects a watch file-size threshold above the parser ceiling', async () => {
await expect(
resolveWatchOptions(
repoPath,
{ maxFileSize: '32769' },
{
maxFileSize: undefined,
workerTimeout: undefined,
verbose: undefined,
},
),
).rejects.toThrow('maxFileSize must not exceed 32768');
});
it.skipIf(process.platform === 'win32')(
'rejects repository ignore files that are final-file symlinks',
async () => {
const outside = await fs.mkdtemp(path.join(os.tmpdir(), 'gitnexus-watch-outside-'));
try {
const target = path.join(outside, 'ignore');
await fs.writeFile(target, 'secret.ts\n');
await fs.symlink(target, path.join(repoPath, '.gitignore'), 'file');
await expect(createWatchIgnorePredicate(repoPath)).rejects.toThrow(/symbolic link/);
} finally {
await fs.rm(outside, { recursive: true, force: true });
}
},
);
});

View file

@ -0,0 +1,348 @@
import { afterEach, describe, expect, it, vi } from 'vitest';
import { WATCH_FULL_REFRESH_PATH, WatchRefreshQueue } from '../../src/cli/watch-queue.js';
afterEach(() => vi.useRealTimers());
describe('WatchRefreshQueue', () => {
it('propagates an initial refresh failure without reporting it as retryable', async () => {
const onError = vi.fn();
const queue = new WatchRefreshQueue(
async () => {
throw new Error('initial analyze failed');
},
onError,
10,
);
await expect(queue.runInitial()).rejects.toThrow('initial analyze failed');
expect(onError).not.toHaveBeenCalled();
await queue.close();
});
it('debounces and deduplicates rapid writes', async () => {
vi.useFakeTimers();
const batches: readonly string[][] = [];
const mutable = batches as string[][];
const queue = new WatchRefreshQueue(
async (paths) => mutable.push([...paths]),
() => {},
100,
);
queue.enqueue('src/a.ts');
queue.enqueue('src/a.ts');
await vi.advanceTimersByTimeAsync(50);
queue.enqueue('src/b.ts');
await vi.advanceTimersByTimeAsync(99);
expect(batches).toEqual([]);
await vi.advanceTimersByTimeAsync(1);
await queue.waitForIdle();
expect(batches).toEqual([['src/a.ts', 'src/b.ts']]);
});
it('queues edits made during a refresh and never overlaps writers', async () => {
vi.useFakeTimers();
let releaseFirst!: () => void;
let active = 0;
let peak = 0;
const batches: string[][] = [];
const queue = new WatchRefreshQueue(
async (paths) => {
active++;
peak = Math.max(peak, active);
batches.push([...paths]);
if (batches.length === 1) await new Promise<void>((resolve) => (releaseFirst = resolve));
active--;
},
() => {},
100,
);
queue.enqueue('src/a.ts');
await vi.advanceTimersByTimeAsync(100);
queue.enqueue('src/b.ts');
queue.enqueue('src/c.ts');
releaseFirst();
await Promise.resolve();
await vi.advanceTimersByTimeAsync(100);
await queue.waitForIdle();
expect(peak).toBe(1);
expect(batches).toEqual([['src/a.ts'], ['src/b.ts', 'src/c.ts']]);
});
it('retries a failed refresh without dropping its batch', async () => {
vi.useFakeTimers();
const errors: string[][] = [];
const successful: string[][] = [];
let attempts = 0;
const queue = new WatchRefreshQueue(
async (paths) => {
attempts++;
if (attempts === 1) throw new Error('failed');
successful.push([...paths]);
},
(_error, paths) => errors.push([...paths]),
10,
);
queue.enqueue('src/a.ts');
await vi.advanceTimersByTimeAsync(10);
expect(errors).toEqual([['src/a.ts']]);
expect(successful).toEqual([]);
await vi.advanceTimersByTimeAsync(250);
await queue.waitForIdle();
expect(errors).toEqual([['src/a.ts']]);
expect(successful).toEqual([['src/a.ts']]);
});
it('parks pre-ready events until the initial refresh completes', async () => {
vi.useFakeTimers();
const batches: string[][] = [];
const queue = new WatchRefreshQueue(
async (paths) => batches.push([...paths]),
() => {},
50,
{ holdEventsUntilInitialRefresh: true },
);
queue.enqueue('src/during-walk.ts');
await vi.advanceTimersByTimeAsync(80);
expect(batches).toEqual([]);
await queue.runInitial();
expect(batches).toEqual([[]]);
await vi.advanceTimersByTimeAsync(50);
await queue.waitForIdle();
expect(batches).toEqual([[], ['src/during-walk.ts']]);
});
it('backs off repeated failures instead of spinning at the debounce interval', async () => {
vi.useFakeTimers();
let attempts = 0;
const queue = new WatchRefreshQueue(
async () => {
attempts++;
if (attempts < 4) throw new Error('still unavailable');
},
() => {},
10,
{ retryBaseDelayMs: 100, retryMaxDelayMs: 400 },
);
queue.enqueue('src/a.ts');
await vi.advanceTimersByTimeAsync(10);
expect(attempts).toBe(1);
await vi.advanceTimersByTimeAsync(99);
expect(attempts).toBe(1);
await vi.advanceTimersByTimeAsync(1);
expect(attempts).toBe(2);
await vi.advanceTimersByTimeAsync(199);
expect(attempts).toBe(2);
await vi.advanceTimersByTimeAsync(1);
expect(attempts).toBe(3);
await vi.advanceTimersByTimeAsync(399);
expect(attempts).toBe(3);
await vi.advanceTimersByTimeAsync(1);
await queue.waitForIdle();
expect(attempts).toBe(4);
});
it('merges an event during retry backoff without shortening the retry delay', async () => {
vi.useFakeTimers();
const batches: string[][] = [];
let attempts = 0;
const queue = new WatchRefreshQueue(
async (paths) => {
attempts++;
if (attempts === 1) throw new Error('failed');
batches.push([...paths]);
},
() => {},
10,
{ retryBaseDelayMs: 1_000 },
);
queue.enqueue('src/a.ts');
await vi.advanceTimersByTimeAsync(10);
expect(attempts).toBe(1);
await vi.advanceTimersByTimeAsync(100);
queue.enqueue('src/b.ts');
await vi.advanceTimersByTimeAsync(899);
expect(attempts).toBe(1);
await vi.advanceTimersByTimeAsync(1);
await queue.waitForIdle();
expect(attempts).toBe(2);
expect(batches).toEqual([['src/a.ts', 'src/b.ts']]);
});
it('contains a throwing error reporter for a detached refresh', async () => {
vi.useFakeTimers();
const queue = new WatchRefreshQueue(
async () => {
throw new Error('refresh failed');
},
async () => {
throw new Error('reporting failed');
},
10,
);
queue.enqueue('src/a.ts');
await vi.advanceTimersByTimeAsync(10);
await queue.close();
await expect(queue.waitForIdle()).resolves.toBeUndefined();
});
it('contains a synchronously throwing refresh and retries its batch', async () => {
vi.useFakeTimers();
const errors: string[][] = [];
const successful: string[][] = [];
let attempts = 0;
const queue = new WatchRefreshQueue(
(paths) => {
attempts++;
if (attempts === 1) throw new Error('synchronous refresh failure');
successful.push([...paths]);
return Promise.resolve();
},
(_error, paths) => errors.push([...paths]),
10,
);
queue.enqueue('src/a.ts');
await vi.advanceTimersByTimeAsync(10);
expect(errors).toEqual([['src/a.ts']]);
await vi.advanceTimersByTimeAsync(250);
await queue.waitForIdle();
expect(attempts).toBe(2);
expect(successful).toEqual([['src/a.ts']]);
});
it('closes cleanly when a refresh failure triggers shutdown', async () => {
vi.useFakeTimers();
let closePromise: Promise<void> | undefined;
const queue = new WatchRefreshQueue(
async () => {
throw new Error('stop after failure');
},
() => {
closePromise = queue.close();
},
10,
);
queue.enqueue('src/a.ts');
await vi.advanceTimersByTimeAsync(10);
expect(closePromise).toBeDefined();
await expect(closePromise).resolves.toBeUndefined();
});
it('flushes by max wait even when writes never become quiet', async () => {
vi.useFakeTimers();
const batches: string[][] = [];
const queue = new WatchRefreshQueue(
async (paths) => batches.push([...paths]),
() => {},
100,
{ maxWaitMs: 250 },
);
queue.enqueue('src/0.ts');
await vi.advanceTimersByTimeAsync(90);
queue.enqueue('src/1.ts');
await vi.advanceTimersByTimeAsync(90);
queue.enqueue('src/2.ts');
await vi.advanceTimersByTimeAsync(70);
await queue.waitForIdle();
expect(batches).toEqual([['src/0.ts', 'src/1.ts', 'src/2.ts']]);
});
it('bounds high-cardinality paths while retaining priority control files', async () => {
vi.useFakeTimers();
const batches: string[][] = [];
const queue = new WatchRefreshQueue(
async (paths) => batches.push([...paths]),
() => {},
10,
{
maxPendingPaths: 2,
isPriorityPath: (filePath) => filePath === '.gitnexusrc',
},
);
for (let index = 0; index < 20; index++) queue.enqueue(`src/${index}.ts`);
queue.enqueue('.gitnexusrc');
await vi.advanceTimersByTimeAsync(10);
await queue.waitForIdle();
expect(batches).toEqual([[WATCH_FULL_REFRESH_PATH, '.gitnexusrc', 'src/1.ts']]);
});
it('bounds a flood of distinct priority paths', async () => {
vi.useFakeTimers();
const batches: string[][] = [];
const queue = new WatchRefreshQueue(
async (paths) => batches.push([...paths]),
() => {},
10,
{
maxPendingPaths: 2,
isPriorityPath: (filePath) => filePath.endsWith('/.gitignore'),
},
);
for (let index = 0; index < 20; index++) queue.enqueue(`packages/${index}/.gitignore`);
await vi.advanceTimersByTimeAsync(10);
await queue.waitForIdle();
expect(batches).toEqual([
[WATCH_FULL_REFRESH_PATH, 'packages/0/.gitignore', 'packages/1/.gitignore'],
]);
});
it('does not report overflow for a duplicate at capacity', async () => {
vi.useFakeTimers();
const batches: string[][] = [];
const queue = new WatchRefreshQueue(
async (paths) => batches.push([...paths]),
() => {},
10,
{ maxPendingPaths: 2 },
);
queue.enqueue('src/a.ts');
queue.enqueue('src/b.ts');
queue.enqueue('src/a.ts');
await vi.advanceTimersByTimeAsync(10);
await queue.waitForIdle();
expect(batches).toEqual([['src/a.ts', 'src/b.ts']]);
});
it('runs a full refresh when the pending-path limit is zero', async () => {
vi.useFakeTimers();
const batches: string[][] = [];
const queue = new WatchRefreshQueue(
async (paths) => batches.push([...paths]),
() => {},
10,
{ maxPendingPaths: 0 },
);
queue.enqueue('src/a.ts');
await vi.advanceTimersByTimeAsync(10);
await queue.waitForIdle();
expect(batches).toEqual([[WATCH_FULL_REFRESH_PATH]]);
});
});