From 51ca6f8c6d1d0293595ad3bf5261cfdadb95cb0b Mon Sep 17 00:00:00 2001 From: Eva Date: Tue, 14 Jul 2026 01:37:29 +0700 Subject: [PATCH 1/2] ci(release): require protected manual publishing --- .github/scripts/check-release-governance.mjs | 75 +++++++++++++++++++ .github/workflows/ci-quality.yml | 1 + .github/workflows/ci.yml | 7 +- .github/workflows/codeql.yml | 5 +- .github/workflows/publish.yml | 18 ++--- .github/workflows/workflow-lint.yml | 6 +- gitnexus/src/cli/doctor.ts | 4 +- gitnexus/src/core/rerank/voyage-provider.ts | 13 ++-- gitnexus/src/mcp/local/local-backend.ts | 10 +-- gitnexus/test/unit/calltool-dispatch.test.ts | 37 +++++---- gitnexus/test/unit/rerank-provider.test.ts | 13 ++-- .../test/unit/voyage-rerank-provider.test.ts | 22 +++--- package-lock.json | 3 +- package.json | 4 +- 14 files changed, 141 insertions(+), 77 deletions(-) create mode 100644 .github/scripts/check-release-governance.mjs diff --git a/.github/scripts/check-release-governance.mjs b/.github/scripts/check-release-governance.mjs new file mode 100644 index 0000000000..a351620b72 --- /dev/null +++ b/.github/scripts/check-release-governance.mjs @@ -0,0 +1,75 @@ +#!/usr/bin/env node + +import fs from 'node:fs'; +import path from 'node:path'; +import { fileURLToPath } from 'node:url'; +import { parse } from 'yaml'; + +const repoRoot = path.resolve(path.dirname(fileURLToPath(import.meta.url)), '../..'); +const readWorkflow = (filename) => + parse(fs.readFileSync(path.join(repoRoot, '.github/workflows', filename), 'utf8')); +const workflow = readWorkflow('publish.yml'); + +const fail = (message) => { + process.stderr.write(`release governance check failed: ${message}\n`); + process.exitCode = 1; +}; + +const triggers = workflow.on; +if (!triggers || typeof triggers !== 'object') { + fail('publish.yml must declare structured triggers'); +} else { + const push = triggers.push; + if (!push || typeof push !== 'object') { + fail('publish.yml must retain stable tag publishing'); + } else { + if ('branches' in push || 'branches-ignore' in push) { + fail('publish.yml must not publish from branch pushes'); + } + const tags = Array.isArray(push.tags) ? push.tags : []; + if (!tags.includes('v*') || !tags.includes('!v*-rc.*')) { + fail('publish.yml must accept stable tags and reject self-produced RC tags'); + } + } + if (!('workflow_dispatch' in triggers)) { + fail('publish.yml must retain an explicit manual RC trigger'); + } + if ('pull_request' in triggers || 'pull_request_target' in triggers) { + fail('publish.yml must never publish from pull request events'); + } +} + +const publishJob = workflow.jobs?.publish; +const environment = + typeof publishJob?.environment === 'string' + ? publishJob.environment + : publishJob?.environment?.name; +if (environment !== 'internal-release') { + fail('the publish job must require the internal-release environment'); +} + +for (const filename of [ + 'ci.yml', + 'codeql.yml', + 'gitleaks.yml', + 'dependency-review.yml', + 'workflow-lint.yml', +]) { + const candidate = readWorkflow(filename); + const pullRequest = candidate.on?.pull_request; + if (!pullRequest || typeof pullRequest !== 'object') { + fail(`${filename} must run for pull requests`); + continue; + } + const branches = Array.isArray(pullRequest.branches) ? pullRequest.branches : []; + if (!branches.includes('main')) { + fail(`${filename} must run for main-targeted pull requests`); + } + if ('paths' in pullRequest || 'paths-ignore' in pullRequest) { + fail(`${filename} must not omit required checks through path filters`); + } +} + +if (!process.exitCode) { + process.stdout.write('release governance check passed\n'); +} diff --git a/.github/workflows/ci-quality.yml b/.github/workflows/ci-quality.yml index 0106049ae3..069e44e82d 100644 --- a/.github/workflows/ci-quality.yml +++ b/.github/workflows/ci-quality.yml @@ -20,6 +20,7 @@ jobs: cache: npm cache-dependency-path: package-lock.json - run: npm ci + - run: npm run check:release-governance - run: npx prettier --check . lint: diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 36421d8890..ec32cc96db 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -3,7 +3,6 @@ name: CI on: pull_request: branches: [main, 'reconcile/**'] - paths-ignore: ['**.md', 'docs/**', 'LICENSE'] workflow_call: permissions: @@ -16,9 +15,9 @@ permissions: # prefix that could resolve to the caller's name would share a concurrency group # with the caller → deadlock. A literal prefix is immune. Direct `pull_request` # invocations use `CI-`; invocations from a reusable-workflow caller fall -# into a per-run-unique group that never serializes with the caller. `push` to -# main is handled by publish.yml (RC mode), which calls this workflow once -# before publishing. +# into a per-run-unique group that never serializes with the caller. A manual +# RC dispatch or stable tag in publish.yml calls this workflow once before the +# protected release job can begin. concurrency: group: ${{ github.event_name == 'pull_request' && format('CI-{0}', github.ref) || format('CI-nested-{0}', github.run_id) }} cancel-in-progress: ${{ github.event_name == 'pull_request' }} diff --git a/.github/workflows/codeql.yml b/.github/workflows/codeql.yml index 04d6d8dd26..4c7b7aa53d 100644 --- a/.github/workflows/codeql.yml +++ b/.github/workflows/codeql.yml @@ -3,13 +3,12 @@ name: CodeQL # Static analysis (SAST) for TypeScript/JavaScript and Python sources. # Findings upload to the GitHub Security tab as SARIF. # -# Advisory only on first introduction — see docs/plans/2026-05-03-001-feat-automated-security-scans-plan.md. -# Promote to a required check after baseline triage (operator decision). +# Required on every main-targeted pull request. Scheduled and main-push runs +# continue to catch findings that appear between pull requests. on: pull_request: branches: [main] - paths-ignore: ['**.md', 'docs/**', 'LICENSE'] push: branches: [main] schedule: diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml index 62679fe7c1..fb9d452883 100644 --- a/.github/workflows/publish.yml +++ b/.github/workflows/publish.yml @@ -6,7 +6,7 @@ name: Publish # double-publish race this unification closes. # # Two release modes, both routed through this file: -# • Release candidate (rc) — triggered by push to `main` or workflow_dispatch. +# • Release candidate (rc) — triggered only by workflow_dispatch on `main`. # The RC path computes the next rc version, applies it in-CI, pushes a # detached release commit with v-rc. + rc/ marker # atomically, then publishes to npm with --tag rc and creates a GitHub @@ -28,11 +28,6 @@ name: Publish on: push: - branches: [main] - paths-ignore: - - '**.md' - - 'docs/**' - - 'LICENSE' tags: # Negative-globbed exclusion of RC tags this workflow itself produces # (see the SELF-TRIGGER INVARIANT in the header comment). @@ -65,9 +60,9 @@ on: # Workflow-level deny-all; each job declares the minimum it needs. permissions: {} -# Distinct refs (refs/heads/main, refs/tags/v*) run in parallel. The -# release-PR-skip in rc-guard is the load-bearing invariant that prevents -# an RC main-push and a stable tag-push colliding on the same release commit. +# Manual RC dispatches and stable tag pushes use distinct refs and may run in +# parallel. The release-PR skip remains a defense against a manually dispatched +# RC colliding with a stable tag on the same release commit. concurrency: group: ${{ github.workflow }}-${{ github.ref }} cancel-in-progress: false @@ -118,9 +113,6 @@ jobs: ;; push) case "${GH_REF}" in - refs/heads/main) - MODE="rc" - ;; refs/tags/v*) # The trigger filter already excluded v*-rc.* tags. Anything # reaching here is either a stable semver or a malformed v*. @@ -280,6 +272,8 @@ jobs: if: ${{ always() && needs.ci.result == 'success' && (needs.route.outputs.mode == 'stable' || needs.rc-guard.outputs.should_run == 'true') }} runs-on: ubuntu-latest timeout-minutes: 20 + environment: + name: internal-release permissions: # contents: write — RC path needs it for `git push --atomic` (v-tag + # marker). Stable path runs in the same job and inherits the grant; it diff --git a/.github/workflows/workflow-lint.yml b/.github/workflows/workflow-lint.yml index 302ba9d597..2cec04abaf 100644 --- a/.github/workflows/workflow-lint.yml +++ b/.github/workflows/workflow-lint.yml @@ -6,14 +6,12 @@ name: Workflow Lint # - zizmor: security misconfigurations — unpinned actions, dangerous # `${{ }}` interpolation, missing per-job permissions, etc. # -# Scoped to PRs that touch .github/** only — keeps off the typical PR -# critical path. +# Runs on every main-targeted PR because both jobs are required merge gates. +# Path filtering would leave unrelated PRs waiting on checks that never start. on: pull_request: branches: [main] - paths: - - '.github/**' permissions: contents: read diff --git a/gitnexus/src/cli/doctor.ts b/gitnexus/src/cli/doctor.ts index 3e60e63a86..0163b9f9f7 100644 --- a/gitnexus/src/cli/doctor.ts +++ b/gitnexus/src/cli/doctor.ts @@ -162,9 +162,7 @@ export function repoVectorDoctorStatus(meta: RepoMeta | null | undefined): { const rawEmbeddingCount = Number(meta.stats?.embeddings ?? 0); const embeddingCount = - Number.isFinite(rawEmbeddingCount) && rawEmbeddingCount > 0 - ? Math.floor(rawEmbeddingCount) - : 0; + Number.isFinite(rawEmbeddingCount) && rawEmbeddingCount > 0 ? Math.floor(rawEmbeddingCount) : 0; if (embeddingCount <= 0) return { status: 'no embeddings', detail: null }; const vector = meta.capabilities?.vectorSearch; diff --git a/gitnexus/src/core/rerank/voyage-provider.ts b/gitnexus/src/core/rerank/voyage-provider.ts index 1e9aa9f5f4..48b21344c6 100644 --- a/gitnexus/src/core/rerank/voyage-provider.ts +++ b/gitnexus/src/core/rerank/voyage-provider.ts @@ -1,8 +1,4 @@ -import { - CircuitOpenError, - ResilientFetchExhaustedError, - resilientFetch, -} from 'gitnexus-shared'; +import { CircuitOpenError, ResilientFetchExhaustedError, resilientFetch } from 'gitnexus-shared'; import { type RerankFailurePolicy, @@ -135,7 +131,9 @@ export class VoyageRerankProvider implements RerankProvider { error instanceof DOMException && (error.name === 'TimeoutError' || error.name === 'AbortError') ) { - const action = request.signal?.aborted ? 'cancelled' : `timed out after ${this.config.timeoutMs}ms`; + const action = request.signal?.aborted + ? 'cancelled' + : `timed out after ${this.config.timeoutMs}ms`; throw new Error(`Rerank request ${action} (${safeUrl(url)})`); } if (error instanceof ResilientFetchExhaustedError) { @@ -163,7 +161,8 @@ export class VoyageRerankProvider implements RerankProvider { if (!payload || typeof payload !== 'object') { throw new Error(`Rerank endpoint returned an unexpected response shape (${safeUrl(url)})`); } - const source = (payload as { data?: unknown; results?: unknown }).data ?? + const source = + (payload as { data?: unknown; results?: unknown }).data ?? (payload as { results?: unknown }).results; if (!Array.isArray(source)) { throw new Error(`Rerank endpoint returned an unexpected response shape (${safeUrl(url)})`); diff --git a/gitnexus/src/mcp/local/local-backend.ts b/gitnexus/src/mcp/local/local-backend.ts index 3aa7d39898..6fd1ab45d0 100644 --- a/gitnexus/src/mcp/local/local-backend.ts +++ b/gitnexus/src/mcp/local/local-backend.ts @@ -63,10 +63,7 @@ import { isVectorExtensionSupportedByPlatform, } from '../../core/platform/capabilities.js'; import { PhaseTimer } from '../../core/search/phase-timer.js'; -import { - rerankDocuments, - type RerankRuntime, -} from '../../core/rerank/provider.js'; +import { rerankDocuments, type RerankRuntime } from '../../core/rerank/provider.js'; import { resolveRerankRuntime } from '../../core/rerank/voyage-provider.js'; import { ftsDegradedWarning } from '../../core/search/fts-indexes.js'; import { @@ -742,8 +739,9 @@ export class LocalBackend { private warnedMissingEmbeddingStack = false; constructor( - private readonly rerankRuntimeResolver: (repoName: string) => RerankRuntime | null = - resolveRerankRuntime, + private readonly rerankRuntimeResolver: ( + repoName: string, + ) => RerankRuntime | null = resolveRerankRuntime, ) {} /** diff --git a/gitnexus/test/unit/calltool-dispatch.test.ts b/gitnexus/test/unit/calltool-dispatch.test.ts index ccb92488c6..bdedf8011f 100644 --- a/gitnexus/test/unit/calltool-dispatch.test.ts +++ b/gitnexus/test/unit/calltool-dispatch.test.ts @@ -440,10 +440,7 @@ describe('LocalBackend.callTool', () => { const result = await rerankingBackend.callTool('query', { search_query: 'needle' }); - expect(result.definitions.map((item: { name: string }) => item.name)).toEqual([ - 'b.ts', - 'a.ts', - ]); + expect(result.definitions.map((item: { name: string }) => item.name)).toEqual(['b.ts', 'a.ts']); expect(runtime.provider.rerank).toHaveBeenCalledTimes(1); }); @@ -456,12 +453,14 @@ describe('LocalBackend.callTool', () => { ], ftsAvailable: true, }); - const resolver = vi.fn((): RerankRuntime => ({ - provider: { id: 'unused-test', rerank: vi.fn() }, - candidates: 2, - maxDocChars: 1000, - failurePolicy: 'fallback', - })); + const resolver = vi.fn( + (): RerankRuntime => ({ + provider: { id: 'unused-test', rerank: vi.fn() }, + candidates: 2, + maxDocChars: 1000, + failurePolicy: 'fallback', + }), + ); const bypassBackend = new LocalBackend(resolver); await bypassBackend.init(); @@ -470,10 +469,7 @@ describe('LocalBackend.callTool', () => { rerank: false, }); - expect(result.definitions.map((item: { name: string }) => item.name)).toEqual([ - 'a.ts', - 'b.ts', - ]); + expect(result.definitions.map((item: { name: string }) => item.name)).toEqual(['a.ts', 'b.ts']); expect(resolver).not.toHaveBeenCalled(); expect(result.timing).not.toHaveProperty('rerank'); expect(result.warning).toBeUndefined(); @@ -501,10 +497,7 @@ describe('LocalBackend.callTool', () => { const result = await fallbackBackend.callTool('query', { search_query: 'needle' }); - expect(result.definitions.map((item: { name: string }) => item.name)).toEqual([ - 'a.ts', - 'b.ts', - ]); + expect(result.definitions.map((item: { name: string }) => item.name)).toEqual(['a.ts', 'b.ts']); expect(result.warning).toMatch(/rerank unavailable.*existing bm25\/vector ranking/i); }); @@ -712,9 +705,13 @@ describe('LocalBackend.callTool', () => { expect( cap .records() - .some((record) => /exact scan refused.*2 chunks exceed.*limit of 1/i.test(String(record.msg))), + .some((record) => + /exact scan refused.*2 chunks exceed.*limit of 1/i.test(String(record.msg)), + ), ).toBe(true); - const queries = (executeQuery as any).mock.calls.map(([, cypher]: [string, string]) => cypher); + const queries = (executeQuery as any).mock.calls.map( + ([, cypher]: [string, string]) => cypher, + ); expect(queries.some((cypher: string) => cypher.includes('e.embedding AS embedding'))).toBe( false, ); diff --git a/gitnexus/test/unit/rerank-provider.test.ts b/gitnexus/test/unit/rerank-provider.test.ts index aa969a48c0..aa749d3985 100644 --- a/gitnexus/test/unit/rerank-provider.test.ts +++ b/gitnexus/test/unit/rerank-provider.test.ts @@ -1,9 +1,6 @@ import { describe, expect, it, vi } from 'vitest'; -import { - rerankDocuments, - type RerankProvider, -} from '../../src/core/rerank/provider.js'; +import { rerankDocuments, type RerankProvider } from '../../src/core/rerank/provider.js'; const documents = ['first', 'second', 'third']; @@ -28,7 +25,13 @@ describe('rerankDocuments', () => { it.each([ ['non-array response', null], ['out-of-range index', [{ index: 3, score: 0.5 }]], - ['duplicate index', [{ index: 1, score: 0.5 }, { index: 1, score: 0.4 }]], + [ + 'duplicate index', + [ + { index: 1, score: 0.5 }, + { index: 1, score: 0.4 }, + ], + ], ['non-finite score', [{ index: 0, score: Number.NaN }]], ])('rejects malformed provider output: %s', async (_label, output) => { const provider = { diff --git a/gitnexus/test/unit/voyage-rerank-provider.test.ts b/gitnexus/test/unit/voyage-rerank-provider.test.ts index ca1a3c738d..81e97b64ee 100644 --- a/gitnexus/test/unit/voyage-rerank-provider.test.ts +++ b/gitnexus/test/unit/voyage-rerank-provider.test.ts @@ -60,10 +60,7 @@ describe('resolveRerankRuntime', () => { it.each([ [{ GITNEXUS_RERANK_PROVIDER: 'unknown', GITNEXUS_RERANK_ALLOWED_REPOS: 'repo' }, /provider/i], - [ - { GITNEXUS_RERANK_PROVIDER: 'voyage', GITNEXUS_RERANK_ALLOWED_REPOS: 'repo' }, - /api key/i, - ], + [{ GITNEXUS_RERANK_PROVIDER: 'voyage', GITNEXUS_RERANK_ALLOWED_REPOS: 'repo' }, /api key/i], [ { GITNEXUS_RERANK_PROVIDER: 'voyage', @@ -88,17 +85,20 @@ describe('VoyageRerankProvider', () => { ); const provider = new VoyageRerankProvider(baseConfig, fetchImpl as typeof fetch); - await expect( - provider.rerank({ query: 'needle', documents: ['a', 'b'] }), - ).resolves.toEqual([{ index: 1, score: 0.8 }]); + await expect(provider.rerank({ query: 'needle', documents: ['a', 'b'] })).resolves.toEqual([ + { index: 1, score: 0.8 }, + ]); expect(fetchImpl).toHaveBeenCalledTimes(1); }); it('surfaces timeout without retrying', async () => { - const fetchImpl = vi.fn((_url: string | URL, init?: RequestInit) => - new Promise((_resolve, reject) => { - init?.signal?.addEventListener('abort', () => reject(init.signal?.reason), { once: true }); - }), + const fetchImpl = vi.fn( + (_url: string | URL, init?: RequestInit) => + new Promise((_resolve, reject) => { + init?.signal?.addEventListener('abort', () => reject(init.signal?.reason), { + once: true, + }); + }), ); const provider = new VoyageRerankProvider( { ...baseConfig, timeoutMs: 5 }, diff --git a/package-lock.json b/package-lock.json index 0969cc7a73..da02f5a889 100644 --- a/package-lock.json +++ b/package-lock.json @@ -15,7 +15,8 @@ "husky": "^9.1.7", "lint-staged": "^15.5.0", "prettier": "^3.8.0", - "prettier-plugin-tailwindcss": "^0.7.0" + "prettier-plugin-tailwindcss": "^0.7.0", + "yaml": "2.8.3" } }, "node_modules/@babel/code-frame": { diff --git a/package.json b/package.json index 8a732cb149..fbec2025bf 100644 --- a/package.json +++ b/package.json @@ -5,6 +5,7 @@ "prepare": "husky", "format": "prettier --write .", "format:check": "prettier --check .", + "check:release-governance": "node .github/scripts/check-release-governance.mjs", "lint": "eslint .", "lint:fix": "eslint --fix .", "gitnexus:refresh": "gitnexus analyze --embeddings --skills", @@ -20,7 +21,8 @@ "husky": "^9.1.7", "lint-staged": "^15.5.0", "prettier": "^3.8.0", - "prettier-plugin-tailwindcss": "^0.7.0" + "prettier-plugin-tailwindcss": "^0.7.0", + "yaml": "2.8.3" }, "lint-staged": { "*.{ts,tsx}": [ From 5a617ad3a5f91b3f3ae91dd7d6e88fe65c147e01 Mon Sep 17 00:00:00 2001 From: Eva Date: Tue, 14 Jul 2026 02:19:10 +0700 Subject: [PATCH 2/2] fix(embeddings): resume partial checkpoint windows --- .../src/core/embeddings/embedding-pipeline.ts | 44 +++++++-- gitnexus/src/core/run-analyze.ts | 82 ++++++++++------ gitnexus/src/server/api.ts | 65 +++++++++++-- gitnexus/src/storage/repo-manager.ts | 16 +++- .../test/unit/api-readonly-wiring.test.ts | 21 ++++ gitnexus/test/unit/embedding-pipeline.test.ts | 95 +++++++++++++++++++ gitnexus/test/unit/run-analyze.test.ts | 29 ++++++ 7 files changed, 302 insertions(+), 50 deletions(-) diff --git a/gitnexus/src/core/embeddings/embedding-pipeline.ts b/gitnexus/src/core/embeddings/embedding-pipeline.ts index b40e18e7d1..86e918cb3f 100644 --- a/gitnexus/src/core/embeddings/embedding-pipeline.ts +++ b/gitnexus/src/core/embeddings/embedding-pipeline.ts @@ -311,9 +311,15 @@ export interface EmbeddingPipelineCheckpoint { chunksProcessed: number; } +export interface EmbeddingPipelineCheckpointWindow extends EmbeddingPipelineCheckpoint { + nodeIds: string[]; +} + export interface EmbeddingPipelineOptions { signal?: AbortSignal; checkpointEveryNodes?: number; + forceReembedNodeIds?: ReadonlySet; + onCheckpointWindowStart?: (window: EmbeddingPipelineCheckpointWindow) => Promise; onCheckpoint?: (checkpoint: EmbeddingPipelineCheckpoint) => Promise; } @@ -425,6 +431,7 @@ export const runEmbeddingPipeline = async ( // Phase 2: Query embeddable nodes let nodes = await queryEmbeddableNodes(executeQuery); throwIfCancelled(); + const embeddableNodeIds = new Set(nodes.map((node) => node.id)); // Incremental mode: compare content hashes, delete stale rows, skip fresh ones. // Computed hashes for stale nodes are cached so batchInsertEmbeddings can reuse them @@ -434,16 +441,20 @@ export const runEmbeddingPipeline = async ( // than all up front — see U6 / KTD7. `staleNodeIds` is consulted inside the // batch loop; it stays empty in full (non-incremental) mode so no deletes fire. const staleNodeIds = new Set(); - if (existingEmbeddings && existingEmbeddings.size > 0) { + const forceReembedNodeIds = pipelineOptions.forceReembedNodeIds; + if ( + (existingEmbeddings && existingEmbeddings.size > 0) || + (forceReembedNodeIds && forceReembedNodeIds.size > 0) + ) { const beforeCount = nodes.length; nodes = nodes.filter((n) => { - const existingHash = existingEmbeddings.get(n.id); + const existingHash = existingEmbeddings?.get(n.id); if (existingHash === undefined) { // New node — needs embedding return true; } const currentHash = contentHashForNode(n, finalConfig); - if (currentHash !== existingHash) { + if (currentHash !== existingHash || forceReembedNodeIds?.has(n.id)) { // Content changed — cache hash for reuse during insert, mark for DELETE + re-embed computedStaleHashes.set(n.id, currentHash); staleNodeIds.add(n.id); @@ -460,6 +471,14 @@ export const runEmbeddingPipeline = async ( } } + if (forceReembedNodeIds && forceReembedNodeIds.size > 0) { + const removedPendingNodeIds = [...forceReembedNodeIds].filter( + (nodeId) => !embeddableNodeIds.has(nodeId), + ); + await deleteStaleEmbeddingRows(executeWithReusedStatement, removedPendingNodeIds); + throwIfCancelled(); + } + const totalNodes = nodes.length; if (isDev) { @@ -491,8 +510,11 @@ export const runEmbeddingPipeline = async ( const batchSize = finalConfig.batchSize; const chunkSize = finalConfig.chunkSize; const overlap = finalConfig.overlap; + const checkpointWindowNodeCount = Math.max( + batchSize, + Math.ceil(checkpointEveryNodes / batchSize) * batchSize, + ); let processedNodes = 0; - let lastCheckpointAt = 0; onProgress({ phase: 'embedding', @@ -506,6 +528,17 @@ export const runEmbeddingPipeline = async ( // Process in batches of nodes for (let batchIndex = 0; batchIndex < totalNodes; batchIndex += batchSize) { throwIfCancelled(); + if (pipelineOptions.onCheckpointWindowStart && batchIndex % checkpointWindowNodeCount === 0) { + await pipelineOptions.onCheckpointWindowStart({ + nodesProcessed: processedNodes, + totalNodes, + chunksProcessed: totalChunks, + nodeIds: nodes + .slice(batchIndex, batchIndex + checkpointWindowNodeCount) + .map((node) => node.id), + }); + throwIfCancelled(); + } const batch = nodes.slice(batchIndex, batchIndex + batchSize); // Chunk each node and generate text @@ -631,14 +664,13 @@ export const runEmbeddingPipeline = async ( if ( pipelineOptions.onCheckpoint && - (processedNodes - lastCheckpointAt >= checkpointEveryNodes || processedNodes === totalNodes) + (processedNodes % checkpointWindowNodeCount === 0 || processedNodes === totalNodes) ) { await pipelineOptions.onCheckpoint({ nodesProcessed: processedNodes, totalNodes, chunksProcessed: totalChunks, }); - lastCheckpointAt = processedNodes; throwIfCancelled(); } } diff --git a/gitnexus/src/core/run-analyze.ts b/gitnexus/src/core/run-analyze.ts index debfc073e1..f8622a4bc1 100644 --- a/gitnexus/src/core/run-analyze.ts +++ b/gitnexus/src/core/run-analyze.ts @@ -856,6 +856,7 @@ export async function runFullAnalysis( } let resumeEmbeddingCheckpoint = false; + let pendingEmbeddingNodeIds = new Set(); let embeddingIdentityForRun: EmbeddingIdentity | undefined; if (existingMeta?.embeddingCheckpoint) { if (options.dropEmbeddings) { @@ -876,9 +877,11 @@ export async function runFullAnalysis( ); } resumeEmbeddingCheckpoint = true; + pendingEmbeddingNodeIds = new Set(checkpoint.pendingNodeIds ?? []); log( `Previous analyze ended at an embedding checkpoint ` + - `(${checkpoint.nodesProcessed}/${checkpoint.totalNodes} nodes); resuming from persisted hashes.`, + `(${checkpoint.nodesProcessed}/${checkpoint.totalNodes} nodes); resuming from persisted hashes` + + `${pendingEmbeddingNodeIds.size > 0 ? ` and regenerating ${pendingEmbeddingNodeIds.size} pending node(s)` : ''}.`, ); } } @@ -2005,6 +2008,48 @@ export async function runFullAnalysis( } } + const saveEmbeddingCheckpoint = async ( + checkpoint: { + nodesProcessed: number; + totalNodes: number; + chunksProcessed: number; + }, + pendingNodeIds: string[], + embeddings: number | undefined, + ): Promise => { + const fileHashes: Record = {}; + for (const [key, value] of newFileHashes) fileHashes[key] = value; + await saveMeta(metaDir, { + ...(existingMeta ?? {}), + repoPath, + lastCommit: currentCommit, + indexedAt: new Date().toISOString(), + branch: branchLabel ?? existingMeta?.branch, + remoteUrl: hasGitDir(repoPath) ? getRemoteUrl(repoPath) : undefined, + stats: { + files: pipelineResult.totalFileCount, + nodes: stats.nodes, + edges: stats.edges, + communities: pipelineResult.communityResult?.stats.totalCommunities, + processes: pipelineResult.processResult?.stats.totalProcesses, + embeddings, + }, + schemaVersion: hasGitDir(repoPath) ? INCREMENTAL_SCHEMA_VERSION : undefined, + cjkSegmentation: getSearchFTSCjkSegmentation(), + fileHashes: hasGitDir(repoPath) ? fileHashes : undefined, + cacheKeys: [...parseCache.usedKeys], + incrementalInProgress: undefined, + embeddingCheckpoint: { + at: new Date().toISOString(), + ...checkpoint, + model: embeddingIdentity.model, + dimensions: embeddingIdentity.dimensions, + pendingNodeIds, + }, + pdg: resolvePdgConfig(options), + }); + }; + const embeddingResult = await runEmbeddingPipeline( executeQuery, executeWithReusedStatement, @@ -2022,6 +2067,10 @@ export async function runFullAnalysis( cachedEmbeddingNodeIds.size > 0 ? cachedEmbeddingNodeIds : undefined, existingEmbeddings, { + forceReembedNodeIds: pendingEmbeddingNodeIds, + onCheckpointWindowStart: async ({ nodeIds, ...checkpoint }) => { + await saveEmbeddingCheckpoint(checkpoint, nodeIds, existingMeta?.stats?.embeddings); + }, onCheckpoint: async (checkpoint) => { await checkpointOnce(); const countResult = await executeQuery( @@ -2029,36 +2078,7 @@ export async function runFullAnalysis( ); const countRow = countResult?.[0]; const embeddings = Number(countRow?.cnt ?? countRow?.[0] ?? 0); - const fileHashes: Record = {}; - for (const [key, value] of newFileHashes) fileHashes[key] = value; - await saveMeta(metaDir, { - ...(existingMeta ?? {}), - repoPath, - lastCommit: currentCommit, - indexedAt: new Date().toISOString(), - branch: branchLabel ?? existingMeta?.branch, - remoteUrl: hasGitDir(repoPath) ? getRemoteUrl(repoPath) : undefined, - stats: { - files: pipelineResult.totalFileCount, - nodes: stats.nodes, - edges: stats.edges, - communities: pipelineResult.communityResult?.stats.totalCommunities, - processes: pipelineResult.processResult?.stats.totalProcesses, - embeddings, - }, - schemaVersion: hasGitDir(repoPath) ? INCREMENTAL_SCHEMA_VERSION : undefined, - cjkSegmentation: getSearchFTSCjkSegmentation(), - fileHashes: hasGitDir(repoPath) ? fileHashes : undefined, - cacheKeys: [...parseCache.usedKeys], - incrementalInProgress: undefined, - embeddingCheckpoint: { - at: new Date().toISOString(), - ...checkpoint, - model: embeddingIdentity.model, - dimensions: embeddingIdentity.dimensions, - }, - pdg: resolvePdgConfig(options), - }); + await saveEmbeddingCheckpoint(checkpoint, [], embeddings); }, }, ); diff --git a/gitnexus/src/server/api.ts b/gitnexus/src/server/api.ts index b7e2930cb6..50b461f48d 100644 --- a/gitnexus/src/server/api.ts +++ b/gitnexus/src/server/api.ts @@ -17,6 +17,7 @@ import { canonicalizePath, cloneDirBelongsToEntry, loadMeta, + saveMeta, listRegisteredRepos, getStoragePath, registryPathEquals, @@ -1784,7 +1785,6 @@ export const createServer = async (port: number, host: string = '127.0.0.1') => const embedTimeout = setTimeout(() => { const current = embedJobManager.getJob(job.id); if (current && current.status !== 'complete' && current.status !== 'failed') { - releaseRepoLock(repoLockPath); embedJobManager.cancelJob(job.id, 'Embedding timed out (30 minute limit)'); } }, EMBED_TIMEOUT_MS); @@ -1796,6 +1796,50 @@ export const createServer = async (port: number, host: string = '127.0.0.1') => await withLbugDb(lbugPath, async () => { const { runEmbeddingPipeline } = await import('../core/embeddings/embedding-pipeline.js'); + const [{ getEmbeddingDimensions }, { resolveEmbeddingConfig }] = await Promise.all([ + import('../core/embeddings/embedder.js'), + import('../core/embeddings/config.js'), + ]); + const embeddingIdentity = { + model: process.env.GITNEXUS_EMBEDDING_MODEL ?? resolveEmbeddingConfig().modelId, + dimensions: getEmbeddingDimensions(), + }; + let embeddingMeta = await loadMeta(entry.storagePath); + if (!embeddingMeta) { + throw new Error('Repository metadata is missing; run gitnexus analyze first'); + } + const priorCheckpoint = embeddingMeta.embeddingCheckpoint; + if ( + priorCheckpoint && + (priorCheckpoint.model !== embeddingIdentity.model || + priorCheckpoint.dimensions !== embeddingIdentity.dimensions) + ) { + throw new Error( + `Cannot resume embedding checkpoint: it uses ${priorCheckpoint.model} at ` + + `${priorCheckpoint.dimensions} dimensions, but this run resolves ` + + `${embeddingIdentity.model} at ${embeddingIdentity.dimensions}.`, + ); + } + const forceReembedNodeIds = new Set(priorCheckpoint?.pendingNodeIds ?? []); + const saveEmbeddingCheckpoint = async ( + checkpoint: { + nodesProcessed: number; + totalNodes: number; + chunksProcessed: number; + }, + pendingNodeIds: string[], + ): Promise => { + embeddingMeta = { + ...embeddingMeta, + embeddingCheckpoint: { + at: new Date().toISOString(), + ...checkpoint, + ...embeddingIdentity, + pendingNodeIds, + }, + }; + await saveMeta(entry.storagePath, embeddingMeta); + }; // Fetch existing content hashes for incremental embedding. // Delegated to lbug-adapter which owns the DB query logic and legacy-fallback handling. const { fetchExistingEmbeddingHashes } = await import('../core/lbug/lbug-adapter.js'); @@ -1832,11 +1876,13 @@ export const createServer = async (port: number, host: string = '127.0.0.1') => existingEmbeddings, { signal: embedController.signal, - onCheckpoint: async () => { - // Make each reported pipeline checkpoint durable before the - // next batch starts. A cancelled/restarted job then resumes - // from content hashes already present in LadybugDB. + forceReembedNodeIds, + onCheckpointWindowStart: async ({ nodeIds, ...checkpoint }) => { + await saveEmbeddingCheckpoint(checkpoint, nodeIds); + }, + onCheckpoint: async (checkpoint) => { await flushWAL(); + await saveEmbeddingCheckpoint(checkpoint, []); }, }, ); @@ -1846,18 +1892,16 @@ export const createServer = async (port: number, host: string = '127.0.0.1') => // handles this during process exit, but the server keeps the // connection open for other routes — a CHECKPOINT is enough. await flushWAL(); + embeddingMeta = { ...embeddingMeta, embeddingCheckpoint: undefined }; + await saveMeta(entry.storagePath, embeddingMeta); }); - clearTimeout(embedTimeout); - releaseRepoLock(repoLockPath); // Don't overwrite 'failed' if the job was cancelled while the pipeline was running const current = embedJobManager.getJob(job.id); if (!current || current.status !== 'failed') { embedJobManager.updateJob(job.id, { status: 'complete' }); } } catch (err: any) { - clearTimeout(embedTimeout); - releaseRepoLock(repoLockPath); const current = embedJobManager.getJob(job.id); if (!current || current.status !== 'failed') { embedJobManager.updateJob(job.id, { @@ -1865,6 +1909,9 @@ export const createServer = async (port: number, host: string = '127.0.0.1') => error: err.message || 'Embedding generation failed', }); } + } finally { + clearTimeout(embedTimeout); + releaseRepoLock(repoLockPath); } })(); diff --git a/gitnexus/src/storage/repo-manager.ts b/gitnexus/src/storage/repo-manager.ts index 6839eebc4e..942f45ddb2 100644 --- a/gitnexus/src/storage/repo-manager.ts +++ b/gitnexus/src/storage/repo-manager.ts @@ -200,10 +200,11 @@ export interface RepoMeta { droppedImporterChunks?: number; }; /** - * Durable marker written after a completed embedding batch and LadybugDB - * checkpoint. The graph is valid, but embedding generation has not reached - * final metadata registration. A matching runtime resumes from persisted - * content hashes; a model or dimension mismatch fails before mutation. + * Durable embedding-resume marker. Before a bounded write window begins, + * `pendingNodeIds` records every node that could become partially persisted; + * after the LadybugDB checkpoint it is cleared while progress is retained. + * A matching runtime resumes from persisted hashes and regenerates pending + * nodes; a model or dimension mismatch fails before mutation. */ embeddingCheckpoint?: { at: string; @@ -212,6 +213,13 @@ export interface RepoMeta { chunksProcessed: number; model: string; dimensions: number; + /** + * Nodes in the current checkpoint window. Any of these may have only a + * subset of their chunks persisted after an abrupt process termination, + * so resume must delete and regenerate them even when a persisted row has + * the current content hash. + */ + pendingNodeIds?: string[]; }; /** * Name of the git branch this index represents (#2106). Absent for the diff --git a/gitnexus/test/unit/api-readonly-wiring.test.ts b/gitnexus/test/unit/api-readonly-wiring.test.ts index 7bb8681dc6..8e86e7f31d 100644 --- a/gitnexus/test/unit/api-readonly-wiring.test.ts +++ b/gitnexus/test/unit/api-readonly-wiring.test.ts @@ -57,4 +57,25 @@ describe('api read-only endpoint wiring', () => { expect(embedSection[0]).not.toMatch(/readOnly:\s*true/); } }); + + it('/api/embed keeps the repository lock until cancelled work actually stops', async () => { + const source = await readSource(); + const timeoutSection = source.match( + /const embedTimeout = setTimeout\([\s\S]*?\/\/ Run embedding pipeline asynchronously/, + ); + expect(timeoutSection).not.toBeNull(); + expect(timeoutSection?.[0]).not.toContain('releaseRepoLock(repoLockPath)'); + }); + + it('/api/embed persists and resumes bounded pending windows', async () => { + const source = await readSource(); + const embedSection = source.match( + /\/\/ Run embedding pipeline asynchronously[\s\S]*?res\.status\(202\)/, + ); + expect(embedSection).not.toBeNull(); + expect(embedSection?.[0]).toContain('forceReembedNodeIds'); + expect(embedSection?.[0]).toContain('onCheckpointWindowStart'); + expect(embedSection?.[0]).toContain('pendingNodeIds'); + expect(embedSection?.[0]).toContain('saveMeta'); + }); }); diff --git a/gitnexus/test/unit/embedding-pipeline.test.ts b/gitnexus/test/unit/embedding-pipeline.test.ts index 68ba7fd4b2..8a874e47eb 100644 --- a/gitnexus/test/unit/embedding-pipeline.test.ts +++ b/gitnexus/test/unit/embedding-pipeline.test.ts @@ -709,6 +709,101 @@ describe('runEmbeddingPipeline incremental filter', () => { expect(resumedIds).toEqual([second.id]); }); + it('re-embeds a pending-window node even when its persisted content hash matches', async () => { + mockEmbedderSetup(); + const node = makeNode({ + id: 'Function:pending:src/pending.ts', + name: 'pending', + filePath: 'src/pending.ts', + }); + const currentHash = contentHashForNode(node, DEFAULT_EMBEDDING_CONFIG); + const executeQuery = mockExecuteQuery([node]); + const executeWithReusedStatement = mockExecuteWithReusedStatement(); + const { runEmbeddingPipeline } = + await import('../../src/core/embeddings/embedding-pipeline.js'); + + await runEmbeddingPipeline( + executeQuery, + executeWithReusedStatement, + onProgress, + {}, + undefined, + new Map([[node.id, currentHash]]), + { forceReembedNodeIds: new Set([node.id]) }, + ); + + const deletedIds = stmtCalls + .filter((call) => call.cypher.includes('DELETE')) + .flatMap((call) => call.params.map((param) => param.nodeId)); + const insertedIds = stmtCalls + .filter((call) => call.cypher.includes('CREATE')) + .flatMap((call) => call.params.map((param) => param.nodeId)); + expect(deletedIds).toContain(node.id); + expect(insertedIds).toContain(node.id); + }); + + it('announces each checkpoint window before mutating any node in that window', async () => { + mockEmbedderSetup(); + const first = makeNode({ id: 'Function:first:src/first.ts', name: 'first' }); + const second = makeNode({ id: 'Function:second:src/second.ts', name: 'second' }); + const third = makeNode({ id: 'Function:third:src/third.ts', name: 'third' }); + const executeQuery = mockExecuteQuery([first, second, third]); + const executeWithReusedStatement = mockExecuteWithReusedStatement(); + const windows: string[][] = []; + const mutationCountsAtWindowStart: number[] = []; + const checkpoints: number[] = []; + const { runEmbeddingPipeline } = + await import('../../src/core/embeddings/embedding-pipeline.js'); + + await runEmbeddingPipeline( + executeQuery, + executeWithReusedStatement, + onProgress, + { batchSize: 1 }, + undefined, + new Map(), + { + checkpointEveryNodes: 2, + onCheckpointWindowStart: async ({ nodeIds }) => { + windows.push(nodeIds); + mutationCountsAtWindowStart.push(stmtCalls.length); + }, + onCheckpoint: async ({ nodesProcessed }) => { + checkpoints.push(nodesProcessed); + }, + }, + ); + + expect(windows).toEqual([[first.id, second.id], [third.id]]); + expect(mutationCountsAtWindowStart).toEqual([0, 2]); + expect(checkpoints).toEqual([2, 3]); + }); + + it('deletes pending-window rows whose node is no longer embeddable', async () => { + mockEmbedderSetup(); + const live = makeNode({ id: 'Function:live:src/live.ts', name: 'live' }); + const removedNodeId = 'Function:removed:src/removed.ts'; + const executeQuery = mockExecuteQuery([live]); + const executeWithReusedStatement = mockExecuteWithReusedStatement(); + const { runEmbeddingPipeline } = + await import('../../src/core/embeddings/embedding-pipeline.js'); + + await runEmbeddingPipeline( + executeQuery, + executeWithReusedStatement, + onProgress, + {}, + undefined, + new Map([[removedNodeId, 'persisted-partial-hash']]), + { forceReembedNodeIds: new Set([removedNodeId]) }, + ); + + const deletedIds = stmtCalls + .filter((call) => call.cypher.includes('DELETE')) + .flatMap((call) => call.params.map((param) => param.nodeId)); + expect(deletedIds).toContain(removedNodeId); + }); + it('deletes only stale nodes — new and unchanged nodes are never deleted (#2333 U6)', async () => { mockEmbedderSetup(); diff --git a/gitnexus/test/unit/run-analyze.test.ts b/gitnexus/test/unit/run-analyze.test.ts index b95f6a00fd..af2c64ad80 100644 --- a/gitnexus/test/unit/run-analyze.test.ts +++ b/gitnexus/test/unit/run-analyze.test.ts @@ -17,6 +17,7 @@ import { } from '../../src/storage/repo-manager.js'; import { taintModelVersion } from '../../src/core/ingestion/taint/typescript-model.js'; import { createTempDir } from '../helpers/test-db.js'; +import { readEmbeddingNodeIds } from '../helpers/embedding-seed.js'; describe('run-analyze module', () => { it('exports runFullAnalysis as a function', async () => { @@ -150,8 +151,36 @@ describe('run-analyze module', () => { const finalized = await loadMeta(storagePath); if (!finalized) throw new Error('expected finalized metadata'); + const [pendingNodeId] = await readEmbeddingNodeIds(tmpRepo.dbPath); + if (!pendingNodeId) throw new Error('expected a persisted embedding node'); await saveMeta(storagePath, { ...finalized, + embeddingCheckpoint: { + at: new Date().toISOString(), + nodesProcessed: 0, + totalNodes: 1, + chunksProcessed: 0, + model: 'test-model', + dimensions: 384, + pendingNodeIds: [pendingNodeId], + }, + }); + fetchMock.mockClear(); + + await runFullAnalysis( + tmpRepo.dbPath, + { skipAgentsMd: true, skipSkills: true }, + { onProgress: () => {} }, + ); + + expect(fetchMock).toHaveBeenCalled(); + expect((await loadMeta(storagePath))?.embeddingCheckpoint).toBeUndefined(); + + const resumedPending = await loadMeta(storagePath); + if (!resumedPending) throw new Error('expected pending-window resume metadata'); + fetchMock.mockClear(); + await saveMeta(storagePath, { + ...resumedPending, embeddingCheckpoint: { at: new Date().toISOString(), nodesProcessed: 1,