From e70ac55a5f2d3337ddcbed188b592fa9e75f137f Mon Sep 17 00:00:00 2001 From: Jean-Baptiste THERY Date: Tue, 14 Jul 2026 23:34:24 +0700 Subject: [PATCH] feat(core): make ingestion resumable --- README.md | 18 +- context7.json | 1 + docs/api-reference.md | 9 +- docs/cli-reference.md | 26 +- docs/troubleshooting.md | 10 + llms.txt | 3 + packages/ragmir-core/README.md | 10 +- packages/ragmir-core/src/cli.ts | 91 +-- packages/ragmir-core/src/destroy.test.ts | 25 +- packages/ragmir-core/src/destroy.ts | 13 +- packages/ragmir-core/src/index.ts | 5 + packages/ragmir-core/src/ingest.test.ts | 195 +++++- packages/ragmir-core/src/ingest.ts | 660 ++++++++++++++------ packages/ragmir-core/src/ingestion-state.ts | 305 +++++++++ packages/ragmir-core/src/store.test.ts | 34 + packages/ragmir-core/src/store.ts | 65 +- packages/ragmir-core/src/types.ts | 33 + 17 files changed, 1267 insertions(+), 236 deletions(-) create mode 100644 packages/ragmir-core/src/ingestion-state.ts diff --git a/README.md b/README.md index 4568e95..38a070a 100644 --- a/README.md +++ b/README.md @@ -43,8 +43,10 @@ citation before you propose an edit. ``` `rgr setup` creates ignored local state under `.ragmir/`, installs project-scoped native skills, -and writes local MCP helpers. `rgr ingest` is incremental. The agent receives bounded passages with -the source path, excerpt, chunk, line range, and PDF page when one is available. +and writes local MCP helpers. `rgr ingest` is incremental and commits resumable batches of 25 files +by default. Re-run the same command after an interruption to continue from the last committed +batch. The agent receives bounded passages with the source path, excerpt, chunk, line range, and +PDF page when one is available. Prefer a direct search? Run: @@ -157,6 +159,18 @@ pnpm exec rgr ingest --rebuild The default `local-hash` provider is offline lexical/hash retrieval. Semantic mode uses Transformers.js and requires an explicit model download or a preloaded local model. +### Resume a long ingestion + +```bash +pnpm exec rgr ingest --batch-size 25 +pnpm exec rgr status --json +``` + +Ragmir records per-file progress atomically under ignored `.ragmir/storage/` state. Files from a +committed batch are not parsed or embedded again after a restart. A full `--rebuild` writes to an +isolated generation and activates it only after row and manifest validation, so an interrupted +rebuild leaves the previous searchable index active. + ### Search scanned PDFs ```bash diff --git a/context7.json b/context7.json index a373ecf..f1a3364 100644 --- a/context7.json +++ b/context7.json @@ -26,6 +26,7 @@ "Qwen and Gemma are optional Ragmir Chat profiles, never requirements of Core, the CLI, the TypeScript API, or MCP.", "The `local-hash` embedding provider (default) is a lexical sha256 embedding, not semantic; use `transformers` for semantic retrieval.", "Switching `embeddingProvider` requires `rgr ingest --rebuild`, since the two providers produce incompatible vectors.", + "Ingestion commits resumable file batches under ignored `.ragmir/storage/` state; use `rgr status --json` for progress, and expect `--rebuild` to preserve the active index until a staged generation is validated.", "Run `rgr doctor --fix` after upgrading or misconfiguration to repair scaffolding, `.gitignore` entries, and the agent skill install.", "Config resolves from the caller's working directory (`.ragmir/config.json`), never from the package install path.", "In a monorepo, the nearest configured ancestor is active; use `rgr bases --json` or explicit `--project-root` to verify routing.", diff --git a/docs/api-reference.md b/docs/api-reference.md index 39c4a68..a55be5d 100644 --- a/docs/api-reference.md +++ b/docs/api-reference.md @@ -101,6 +101,7 @@ authentication, authorization, rate limits, and transport security. | Export | Purpose | | --- | --- | | `ingest(options?)` | Incrementally parse, redact, chunk, embed, and store selected files. | +| `getIngestionProgress(config)` | Read durable progress for the latest ingestion run. | | `audit(cwd?)` | Compare files on disk with the current index. | | `previewChunks(options?)` | Return redacted chunks and distributions without writing an index. | | `search(query, options?)` | Return ranked cited passages. | @@ -112,8 +113,10 @@ authentication, authorization, rate limits, and transport security. | `evaluateGoldenQueries(options)` | Score retrieval against a local golden-query file. | `SearchOptions` accepts `cwd`, `topK`, `contextRadius`, `includePaths`, `excludePaths`, -`contextPaths`, `explain`, `signal`, and `timeoutMs`. `IngestOptions`, `ResearchOptions`, and -`ExpandCitationOptions` also accept `signal` and `timeoutMs`. When explanation is enabled, each +`contextPaths`, `explain`, `signal`, and `timeoutMs`. `IngestOptions` also accepts `rebuild`, a +positive `batchSize` that defaults to 25 files, and an optional `onProgress` callback. Its durable +progress contains the run ID, resume flag, last activity, chunk count, and per-stage file counts. +`IngestOptions`, `ResearchOptions`, and `ExpandCitationOptions` accept `signal` and `timeoutMs`. When explanation is enabled, each result includes reciprocal-rank fusion contributions, one-based vector and lexical ranks, vector distance, lexical backend score, and matched query terms. `ExpandCitationOptions.contextRadius` is clamped to three chunks. @@ -182,7 +185,7 @@ types that callers commonly compose explicitly. | Area | Exported types | | --- | --- | | Configuration | `Config`, `PrivacyProfile`, `RetrievalProfile` | -| Ingestion | `IngestOptions`, `IngestResult`, `AuditReport`, `ChunkStats`, `IngestionLimitsReport`, `IndexManifest`, `IndexManifestFile`, `ParsedPage` | +| Ingestion | `IngestOptions`, `IngestResult`, `IngestionProgress`, `IngestionFileStage`, `IngestionRunMode`, `IngestionRunStatus`, `AuditReport`, `ChunkStats`, `IngestionLimitsReport`, `IndexManifest`, `IndexManifestFile`, `ParsedPage` | | Preview | `PreviewChunksOptions`, `PreviewReport`, `PreviewFile`, `PreviewChunk` | | Retrieval | `SearchOptions`, `SearchResult`, `SearchContextChunk`, `SearchScoreExplanation`, `AskResult`, `CompactSearchResult`, `ExpandCitationOptions`, `ExpandedCitation` | | Research and evaluation | `ResearchOptions`, `ResearchReport`, `ResearchEvidence`, `CodeEvidence`, `SourceDiagnostics`, `SourceDuplicateCandidate`, `SourcePathCandidate`, `EvaluationOptions`, `EvaluationResult`, `EvaluationCaseResult`, `GoldenQuery` | diff --git a/docs/cli-reference.md b/docs/cli-reference.md index 1965353..ad524c7 100644 --- a/docs/cli-reference.md +++ b/docs/cli-reference.md @@ -17,13 +17,13 @@ rgr search "release decision" | `init` | Create basic local configuration only. | | `doctor [--fix]` | Check setup, index freshness, and safe repairs. | | `preview` | Parse, redact, and chunk selected sources without writing the index. | -| `ingest [--rebuild]` | Index configured sources; rebuild after provider or chunking changes. | +| `ingest [--rebuild] [--batch-size N]` | Index configured sources in resumable batches; rebuild after provider or chunking changes. | | `search ` | Return ranked cited passages. | | `ask ` | Return cited context without model synthesis. | | `research ` | Run an audit-backed multi-query retrieval pass. | | `audit [--unsupported]` | Compare sources with the index and list skipped files. | | `bases` | List root and nested monorepo bases and mark the active one. | -| `status` | Show configuration and indexed chunk count. | +| `status` | Show configuration, indexed chunk count, and the latest ingestion progress. | | `security-audit [--strict]` | Check local privacy and Git-ignore posture. | ## Sources and retrieval @@ -46,6 +46,28 @@ agent context is limited. `preview` uses the active redaction and chunking configuration but never writes storage. `audit` reports min, mean, p50, p95, and max chunk sizes plus structural-context coverage. +## Resumable ingestion + +```bash +rgr ingest +rgr status --json +rgr ingest --batch-size 10 +``` + +The default batch contains 25 files. After each batch, Ragmir atomically records per-file state and +the current manifest under `.ragmir/storage/`. Starting `rgr ingest` again resumes a compatible +interrupted run and processes only pending, failed, or changed files. Files already committed to +the index are not parsed or embedded again. + +`rgr status --json` exposes the run ID, mode, status, resume flag, last activity, batch size, chunk +count, and file counts for `pending`, `parsed`, `embedded`, `indexed`, and `error` states. The human +output shows the same progress in a compact form. + +`rgr ingest --rebuild` writes batches into an isolated LanceDB generation. The existing index stays +active until the new table and manifest pass row-count, checksum, and duplicate-ID validation. The +final atomic manifest replacement activates the generation. Re-run the command after interruption +to resume the staged generation. + ## Monorepos ```bash diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index e5aa6e8..8cd77bc 100644 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -16,6 +16,16 @@ Run `rgr setup`. It creates `.ragmir/config.json`, local ignore rules, and optio Check `sources` with `rgr sources list`, then run `rgr ingest`. Use `rgr audit` to compare supported files with the index. Use `rgr ingest --rebuild` after changing embedding provider, model, or chunking. +## Ingestion was interrupted + +Run `rgr status --json`, then start `rgr ingest` again. A compatible run resumes from its last +committed file batch. Files in `parsed` or `embedded` state without a committed index write are +retried; files already in `indexed` state are not parsed or embedded again. If source checksums or +the indexing policy changed, Ragmir starts a new safe run instead. + +An interrupted `rgr ingest --rebuild` leaves the previous complete index active. Re-run the rebuild +to continue its isolated generation. + ## A PDF or image has no text `rgr ingest --json` reports `emptyTextFiles`. For scanned PDFs, run: diff --git a/llms.txt b/llms.txt index 0d13c0e..9002748 100644 --- a/llms.txt +++ b/llms.txt @@ -9,6 +9,9 @@ store. - Install: `pnpm add -D @jcode.labs/ragmir` - First use: `pnpm exec rgr setup && pnpm exec rgr ingest` +- Resumable ingestion: `rgr ingest` commits 25 files per durable batch by default. Re-run it after an + interruption, inspect progress with `rgr status --json`, and use `--batch-size` to change the + checkpoint size. Rebuilds stay isolated until final manifest validation and activation. - CLI: `rgr preview`, `rgr search "query" --explain`, `rgr bases`, `rgr doctor`, `rgr audit`, and `rgr security-audit` - Core behavior: cited retrieval only. `ask` returns evidence, while synthesis belongs to the calling diff --git a/packages/ragmir-core/README.md b/packages/ragmir-core/README.md index 4fcab5f..4deb119 100644 --- a/packages/ragmir-core/README.md +++ b/packages/ragmir-core/README.md @@ -84,8 +84,10 @@ npx rgr search "Which decision changed the rollout?" ``` The project owns an ignored `.ragmir/` directory containing configuration and generated local -state. Ingestion is incremental, and every result identifies the source path, chunk, line range, and -PDF page when one is available. +state. Ingestion is incremental and commits 25 files per resumable batch by default. Re-run the +same command after an interruption, or inspect `npx rgr status --json` for the current run ID, +counts, last activity, and resume flag. Every result identifies the source path, chunk, line range, +and PDF page when one is available. ## CLI essentials @@ -107,6 +109,10 @@ npx rgr research "deployment obligations" --compact npx rgr search "deployment decision" --json ``` +Use `npx rgr ingest --batch-size 10` when a smaller durable checkpoint is useful. Full rebuilds use +an isolated local generation and switch the active manifest only after validation, so a failed or +interrupted rebuild does not replace the last healthy index. + `ask` returns cited retrieval context, not LLM synthesis. `research` performs an audit-backed, multi-query retrieval pass and reports missing or weak evidence. diff --git a/packages/ragmir-core/src/cli.ts b/packages/ragmir-core/src/cli.ts index c1cc669..144d66f 100644 --- a/packages/ragmir-core/src/cli.ts +++ b/packages/ragmir-core/src/cli.ts @@ -25,6 +25,7 @@ import { evaluateGoldenQueries } from "./evaluate.js" import { countSkippedByReason } from "./files.js" import { getIndexFreshnessWarning, getLexicalScanWarning } from "./index-diagnostics.js" import { audit, ingest } from "./ingest.js" +import { getIngestionProgress } from "./ingestion-state.js" import { initProject } from "./init.js" import { discoverKnowledgeBases, knowledgeBaseIdentity } from "./knowledge-bases.js" import { ingestionLimits } from "./limits.js" @@ -377,46 +378,53 @@ program .command("ingest") .description("Parse changed documents, redact, chunk, embed locally, and update LanceDB.") .option("--rebuild", "Force a full local index rebuild instead of reusing unchanged rows.") + .option("--batch-size ", "Files committed per resumable batch.", parsePositiveInt) .option("--json", "Print machine-readable JSON.") - .action(async (options: { rebuild?: boolean; json?: boolean }, command: Command) => { - const cwd = projectRoot(command) - const ingestOptions: Parameters[0] = { cwd } - addOption(ingestOptions, "rebuild", options.rebuild) - const result = await ingest(ingestOptions) - if (options.json) { - console.log(JSON.stringify(result, null, 2)) - if (result.errors.length > 0) { - process.exitCode = 1 + .action( + async ( + options: { rebuild?: boolean; batchSize?: number; json?: boolean }, + command: Command, + ) => { + const cwd = projectRoot(command) + const ingestOptions: Parameters[0] = { cwd } + addOption(ingestOptions, "rebuild", options.rebuild) + addOption(ingestOptions, "batchSize", options.batchSize) + const result = await ingest(ingestOptions) + if (options.json) { + console.log(JSON.stringify(result, null, 2)) + if (result.errors.length > 0) { + process.exitCode = 1 + } + return } - return - } - console.log( - pc.green( - `Done. discoveredFiles=${result.discoveredFiles} supportedFiles=${result.supportedFiles} supportedBytes=${result.supportedBytes} largestFileBytes=${result.largestFileBytes} indexedFiles=${result.indexedFiles} rebuiltFiles=${result.rebuiltFiles} reusedFiles=${result.reusedFiles} chunks=${result.chunks} skippedFiles=${result.skippedFiles} unsupportedFiles=${result.unsupportedFiles} oversizedFiles=${result.oversizedFiles} sensitiveFiles=${result.sensitiveFiles} emptyTextFiles=${result.emptyTextFiles.length} redactions=${result.redactions} errors=${result.errors.length}`, - ), - ) - printUnsupportedSummary(result.unsupportedExtensions) - printEmptyTextFiles(result.emptyTextFiles) - if (result.vectorIndexWarning) { - console.log(pc.yellow(result.vectorIndexWarning)) - } - if (result.lexicalIndexWarning) { - console.log(pc.yellow(result.lexicalIndexWarning)) - } - if (result.unsupportedFiles > 0 || result.oversizedFiles > 0 || result.sensitiveFiles > 0) { - const auditCommand = await rgrCommand(cwd, ["audit", "--unsupported"]) console.log( - pc.yellow(`Some files were not indexed. Run \`${auditCommand.display}\` for details.`), + pc.green( + `Done. runId=${result.runId} resumed=${result.resumed} batchSize=${result.batchSize} discoveredFiles=${result.discoveredFiles} supportedFiles=${result.supportedFiles} supportedBytes=${result.supportedBytes} largestFileBytes=${result.largestFileBytes} indexedFiles=${result.indexedFiles} rebuiltFiles=${result.rebuiltFiles} reusedFiles=${result.reusedFiles} chunks=${result.chunks} skippedFiles=${result.skippedFiles} unsupportedFiles=${result.unsupportedFiles} oversizedFiles=${result.oversizedFiles} sensitiveFiles=${result.sensitiveFiles} emptyTextFiles=${result.emptyTextFiles.length} redactions=${result.redactions} errors=${result.errors.length}`, + ), ) - } - for (const error of result.errors) { - console.error(pc.red(` - ${error.path}: ${error.message}`)) - } - if (result.errors.length > 0) { - process.exitCode = 1 - } - }) + printUnsupportedSummary(result.unsupportedExtensions) + printEmptyTextFiles(result.emptyTextFiles) + if (result.vectorIndexWarning) { + console.log(pc.yellow(result.vectorIndexWarning)) + } + if (result.lexicalIndexWarning) { + console.log(pc.yellow(result.lexicalIndexWarning)) + } + if (result.unsupportedFiles > 0 || result.oversizedFiles > 0 || result.sensitiveFiles > 0) { + const auditCommand = await rgrCommand(cwd, ["audit", "--unsupported"]) + console.log( + pc.yellow(`Some files were not indexed. Run \`${auditCommand.display}\` for details.`), + ) + } + for (const error of result.errors) { + console.error(pc.red(` - ${error.path}: ${error.message}`)) + } + if (result.errors.length > 0) { + process.exitCode = 1 + } + }, + ) program .command("preview") @@ -970,6 +978,7 @@ program const config = await loadConfig(cwd) const identity = knowledgeBaseIdentity(config.projectRoot) const rows = await countRows(config) + const ingestion = await getIngestionProgress(config) const status = { knowledgeBaseId: identity?.id ?? null, projectRoot: config.projectRoot, @@ -999,6 +1008,7 @@ program legacyWordCommand: config.legacyWordCommand, legacyWordTimeoutMs: config.legacyWordTimeoutMs, chunksIndexed: rows, + ingestion, } if (options.json) { console.log(JSON.stringify(status, null, 2)) @@ -1033,6 +1043,17 @@ program console.log(`legacyWordCommand=${config.legacyWordCommand.join(" ")}`) console.log(`legacyWordTimeoutMs=${config.legacyWordTimeoutMs}`) console.log(`chunksIndexed=${rows}`) + if (ingestion) { + console.log(`ingestionRunId=${ingestion.runId}`) + console.log(`ingestionStatus=${ingestion.status}`) + console.log(`ingestionMode=${ingestion.mode}`) + console.log(`ingestionResumed=${ingestion.resumed}`) + console.log(`ingestionBatchSize=${ingestion.batchSize}`) + console.log( + `ingestionProgress=${ingestion.indexedFiles}/${ingestion.totalFiles} indexed, ${ingestion.errorFiles} errors, ${ingestion.pendingFiles} pending`, + ) + console.log(`ingestionLastActivityAt=${ingestion.lastActivityAt}`) + } }) program diff --git a/packages/ragmir-core/src/destroy.test.ts b/packages/ragmir-core/src/destroy.test.ts index 4994d0f..0242f60 100644 --- a/packages/ragmir-core/src/destroy.test.ts +++ b/packages/ragmir-core/src/destroy.test.ts @@ -4,6 +4,7 @@ import os from "node:os" import path from "node:path" import { afterEach, describe, expect, it } from "vitest" import { destroyIndex } from "./destroy.js" +import { createIngestionRunState, writeIngestionState } from "./ingestion-state.js" import { testConfig } from "./test-support/config.js" const tempDirs: string[] = [] @@ -39,6 +40,28 @@ describe("destroyIndex", () => { expect(result.removed).toBe(false) }) + it("removes storage from an interrupted first run with valid ingestion state", async () => { + const root = await mkdtemp(path.join(os.tmpdir(), "ragmir-destroy-ingestion-state-")) + tempDirs.push(root) + const config = testConfig(root) + await writeIngestionState( + createIngestionRunState({ + mode: "incremental", + tableName: config.tableName, + previousTableName: null, + policyFingerprint: "test-policy", + batchSize: 25, + files: [], + reusablePaths: new Set(), + reusableChunkCounts: new Map(), + }), + config, + ) + + await expect(destroyIndex(root)).resolves.toMatchObject({ removed: true }) + expect(existsSync(config.storageDir)).toBe(false) + }) + it("writes a destroy-index access log entry", async () => { const root = await mkdtemp(path.join(os.tmpdir(), "ragmir-destroy-log-")) tempDirs.push(root) @@ -67,7 +90,7 @@ describe("destroyIndex", () => { await mkdir(config.storageDir, { recursive: true }) await writeFile(path.join(config.storageDir, "unrelated.txt"), "keep", "utf8") - await expect(destroyIndex(root)).rejects.toThrow("does not contain index-manifest.json") + await expect(destroyIndex(root)).rejects.toThrow("contains neither index-manifest.json") expect(existsSync(config.storageDir)).toBe(true) }) }) diff --git a/packages/ragmir-core/src/destroy.ts b/packages/ragmir-core/src/destroy.ts index 02c0ab3..243df9f 100644 --- a/packages/ragmir-core/src/destroy.ts +++ b/packages/ragmir-core/src/destroy.ts @@ -6,13 +6,15 @@ import { recordAccess } from "./access-log.js" import { loadConfig } from "./config.js" import { INDEX_MANIFEST_FILENAME } from "./defaults.js" import { clearTransformersCache } from "./embeddings.js" +import { readIngestionState } from "./ingestion-state.js" import type { DestroyIndexResult } from "./types.js" export async function destroyIndex(cwd = process.cwd()): Promise { const config = await loadConfig(cwd) const existed = existsSync(config.storageDir) + const hasIngestionState = existed && (await readIngestionState(config)) !== null - await assertSafeIndexStorage(config.projectRoot, config.storageDir, existed) + await assertSafeIndexStorage(config.projectRoot, config.storageDir, existed, hasIngestionState) await recordAccess(config, { action: "destroy-index" }) await rm(config.storageDir, { recursive: true, force: true }) @@ -31,6 +33,7 @@ async function assertSafeIndexStorage( projectRoot: string, storageDir: string, existed: boolean, + hasIngestionState: boolean, ): Promise { const resolvedProjectRoot = await realpath(projectRoot) const resolvedStorageDir = existed ? await realpath(storageDir) : path.resolve(storageDir) @@ -47,9 +50,13 @@ async function assertSafeIndexStorage( ) } - if (existed && !existsSync(path.join(resolvedStorageDir, INDEX_MANIFEST_FILENAME))) { + if ( + existed && + !hasIngestionState && + !existsSync(path.join(resolvedStorageDir, INDEX_MANIFEST_FILENAME)) + ) { throw new Error( - `Refusing to remove ${JSON.stringify(storageDir)} because it does not contain ${INDEX_MANIFEST_FILENAME}.`, + `Refusing to remove ${JSON.stringify(storageDir)} because it contains neither ${INDEX_MANIFEST_FILENAME} nor a valid ingestion state.`, ) } } diff --git a/packages/ragmir-core/src/index.ts b/packages/ragmir-core/src/index.ts index 02f9b35..61a1119 100644 --- a/packages/ragmir-core/src/index.ts +++ b/packages/ragmir-core/src/index.ts @@ -19,6 +19,7 @@ export { INDEX_SCHEMA_VERSION, } from "./index-diagnostics.js" export { audit, ingest } from "./ingest.js" +export { getIngestionProgress } from "./ingestion-state.js" export { initProject } from "./init.js" export { discoverKnowledgeBases, knowledgeBaseIdentity } from "./knowledge-bases.js" export { ingestionLimits } from "./limits.js" @@ -93,7 +94,11 @@ export type { GoldenQuery, IndexManifest, IndexManifestFile, + IngestionFileStage, IngestionLimitsReport, + IngestionProgress, + IngestionRunMode, + IngestionRunStatus, IngestOptions, IngestResult, KnowledgeBaseContextReport, diff --git a/packages/ragmir-core/src/ingest.test.ts b/packages/ragmir-core/src/ingest.test.ts index 914df0d..d9f18bf 100644 --- a/packages/ragmir-core/src/ingest.test.ts +++ b/packages/ragmir-core/src/ingest.test.ts @@ -5,8 +5,9 @@ import { afterEach, describe, expect, it } from "vitest" import { loadConfig } from "./config.js" import { indexPolicyFingerprint } from "./index-policy.js" import { audit, ingest } from "./ingest.js" +import { getIngestionProgress, readIngestionState, writeIngestionState } from "./ingestion-state.js" import { initProject } from "./init.js" -import { openRowsTable, readIndexManifest } from "./store.js" +import { openRowsTable, readIndexManifest, readRows } from "./store.js" const tempDirs: string[] = [] @@ -76,6 +77,198 @@ describe("ingest", () => { expect(second.indexedFiles).toBe(2) expect(second.rebuiltFiles).toBe(1) expect(second.reusedFiles).toBe(1) + const rows = await readRows(await loadConfig(root)) + expect(rows.map((row) => row.relativePath).sort()).toEqual([ + ".ragmir/raw/alpha.md", + ".ragmir/raw/beta.md", + ]) + expect(new Set(rows.map((row) => row.id)).size).toBe(rows.length) + }) + + it("resumes an interrupted run without reprocessing committed files or duplicating chunks", async () => { + const root = await mkdtemp(path.join(os.tmpdir(), "ragmir-resume-")) + tempDirs.push(root) + await initProject(root) + await mkdir(path.join(root, ".ragmir", "raw"), { recursive: true }) + for (const name of ["alpha", "beta", "gamma"]) { + await writeFile( + path.join(root, ".ragmir", "raw", `${name}.md`), + `${name} evidence.\n`, + "utf8", + ) + } + + const controller = new AbortController() + await expect( + ingest({ + cwd: root, + batchSize: 1, + signal: controller.signal, + onProgress(progress) { + if (progress.indexedFiles === 1) { + controller.abort("test interruption") + } + }, + }), + ).rejects.toMatchObject({ code: "ABORTED" }) + + const config = await loadConfig(root) + const interruptedState = await readIngestionState(config) + const committedFile = interruptedState?.files.find((file) => file.state === "indexed") + expect(interruptedState?.status).toBe("interrupted") + expect(committedFile).toBeDefined() + expect(await readRows(config)).toHaveLength(1) + + if (!interruptedState || !committedFile) { + throw new Error("Expected one committed file in the interrupted ingestion state.") + } + await writeIngestionState( + { + ...interruptedState, + files: interruptedState.files.map((file) => + file.relativePath === committedFile.relativePath ? { ...file, state: "embedded" } : file, + ), + }, + config, + ) + + const resumedProgress: number[] = [] + const resumed = await ingest({ + cwd: root, + onProgress(progress) { + resumedProgress.push(progress.indexedFiles) + }, + }) + const completedState = await readIngestionState(config) + const committedAfterResume = completedState?.files.find( + (file) => file.relativePath === committedFile?.relativePath, + ) + const rows = await readRows(config) + + expect(resumed.runId).toBe(interruptedState?.runId) + expect(resumed.resumed).toBe(true) + expect(resumed.batchSize).toBe(1) + expect(resumedProgress[0]).toBe(1) + expect(completedState?.status).toBe("completed") + expect(committedAfterResume?.updatedAt).toBe(committedFile?.updatedAt) + expect(rows).toHaveLength(3) + expect(new Set(rows.map((row) => row.id)).size).toBe(rows.length) + await expect(getIngestionProgress(config)).resolves.toMatchObject({ + runId: resumed.runId, + resumed: true, + indexedFiles: 3, + pendingFiles: 0, + errorFiles: 0, + }) + }) + + it("keeps the active healthy index when a staged rebuild is interrupted", async () => { + const root = await mkdtemp(path.join(os.tmpdir(), "ragmir-rebuild-rollback-")) + tempDirs.push(root) + await initProject(root) + await mkdir(path.join(root, ".ragmir", "raw"), { recursive: true }) + await writeFile( + path.join(root, ".ragmir", "raw", "healthy.md"), + "Healthy production evidence.\n", + "utf8", + ) + await ingest({ cwd: root }) + const config = await loadConfig(root) + const manifestBefore = await readIndexManifest(config) + const rowsBefore = await readRows(config) + await writeFile( + path.join(root, ".ragmir", "raw", "new.md"), + "New evidence for the staged rebuild.\n", + "utf8", + ) + + const controller = new AbortController() + await expect( + ingest({ + cwd: root, + rebuild: true, + batchSize: 1, + signal: controller.signal, + onProgress(progress) { + if (progress.mode === "rebuild" && progress.indexedFiles === 1) { + controller.abort("test rebuild interruption") + } + }, + }), + ).rejects.toMatchObject({ code: "ABORTED" }) + + const manifestAfter = await readIndexManifest(config) + const rowsAfter = await readRows(config) + expect(manifestAfter).toEqual(manifestBefore) + expect(rowsAfter).toEqual(rowsBefore) + await expect(getIngestionProgress(config)).resolves.toMatchObject({ + mode: "rebuild", + status: "interrupted", + indexedFiles: 1, + }) + }) + + it("keeps the active healthy index when a staged rebuild fails", async () => { + const root = await mkdtemp(path.join(os.tmpdir(), "ragmir-rebuild-failure-")) + tempDirs.push(root) + await initProject(root) + await mkdir(path.join(root, ".ragmir", "raw"), { recursive: true }) + await writeFile( + path.join(root, ".ragmir", "raw", "healthy.md"), + "Healthy index before a failed rebuild.\n", + "utf8", + ) + await ingest({ cwd: root }) + const config = await loadConfig(root) + const manifestBefore = await readIndexManifest(config) + const rowsBefore = await readRows(config) + + await expect( + ingest({ + cwd: root, + rebuild: true, + batchSize: 1, + onProgress(progress) { + if (progress.mode === "rebuild" && progress.indexedFiles === 1) { + throw new Error("simulated fatal rebuild failure") + } + }, + }), + ).rejects.toThrow("simulated fatal rebuild failure") + + expect(await readIndexManifest(config)).toEqual(manifestBefore) + expect(await readRows(config)).toEqual(rowsBefore) + await expect(getIngestionProgress(config)).resolves.toMatchObject({ + mode: "rebuild", + status: "failed", + indexedFiles: 1, + }) + }) + + it("continues indexing healthy files when one PDF is corrupt", async () => { + const root = await mkdtemp(path.join(os.tmpdir(), "ragmir-corrupt-pdf-")) + tempDirs.push(root) + await initProject(root) + await mkdir(path.join(root, ".ragmir", "raw"), { recursive: true }) + await writeFile(path.join(root, ".ragmir", "raw", "broken.pdf"), "not a pdf", "utf8") + await writeFile( + path.join(root, ".ragmir", "raw", "healthy.md"), + "Healthy evidence remains indexable.\n", + "utf8", + ) + + const result = await ingest({ cwd: root, batchSize: 1 }) + + expect(result.indexedFiles).toBe(1) + expect(result.errors).toEqual([expect.objectContaining({ path: ".ragmir/raw/broken.pdf" })]) + expect((await readRows(await loadConfig(root))).map((row) => row.relativePath)).toEqual([ + ".ragmir/raw/healthy.md", + ]) + await expect(getIngestionProgress(await loadConfig(root))).resolves.toMatchObject({ + status: "completed_with_errors", + indexedFiles: 1, + errorFiles: 1, + }) }) it("writes an index manifest after ingest and reports a null vectorIndexWarning", async () => { diff --git a/packages/ragmir-core/src/ingest.ts b/packages/ragmir-core/src/ingest.ts index f9537b4..4f25965 100644 --- a/packages/ragmir-core/src/ingest.ts +++ b/packages/ragmir-core/src/ingest.ts @@ -13,25 +13,40 @@ import { import { INDEX_SCHEMA_VERSION } from "./index-diagnostics.js" import { indexPolicyFingerprint } from "./index-policy.js" import { withIndexWriteLock } from "./index-write-lock.js" +import type { IngestionRunState } from "./ingestion-state.js" +import { + canResumeIngestion, + createIngestionRunState, + finishIngestionState, + ingestionProgress, + readIngestionState, + removeStagedIndexManifest, + resumeIngestionState, + updateIngestionFile, + writeIngestionState, + writeStagedIndexManifest, +} from "./ingestion-state.js" import { operationSignal, throwIfAborted } from "./operation.js" import { parseFile } from "./parsing.js" import { redactText, totalRedactions } from "./redaction.js" import { + activeIndexTableName, + dropRowsTable, openRowsTable, + openRowsTableByName, readEmptyTextFiles, readIndexManifest, - updateRows, + updateRowsInTable, writeEmptyTextFiles, writeIndexManifest, - writeRows, } from "./store.js" import type { AuditReport, Config, + IndexManifest, IndexManifestFile, IngestOptions, IngestResult, - RedactionCount, SourceDiagnostics, SourceFile, TextChunk, @@ -40,6 +55,7 @@ import type { import { VERSION } from "./version.js" const MAX_SOURCE_DIAGNOSTIC_ITEMS = 20 +const DEFAULT_INGEST_FILE_BATCH_SIZE = 25 const ARCHIVE_PATH_PATTERNS = [ /(^|[/_-])archive(s)?([/_-]|$)/iu, /(^|[/_-])backup(s)?([/_-]|$)/iu, @@ -77,91 +93,338 @@ async function ingestUnlocked( connection: Connection | undefined, signal: AbortSignal | undefined, ): Promise { - throwIfAborted(signal) - const policyFingerprint = indexPolicyFingerprint(config) - const existingManifest = await readIndexManifest(config) - const manifestCompatible = - !options.rebuild && - existingManifest?.schemaVersion === INDEX_SCHEMA_VERSION && - existingManifest.indexPolicyFingerprint === policyFingerprint && - existingManifest.indexedFiles !== undefined - const existingTable = manifestCompatible ? await openRowsTable(config, connection) : null - const storedEmptyFiles = manifestCompatible ? await readEmptyTextFiles(config) : [] - const knownFiles = new Map( - [...(existingManifest?.indexedFiles ?? []), ...storedEmptyFiles].map((file) => [ - file.relativePath, - file, - ]), - ) - const inventory = await inventorySourceFiles(config, { knownFiles }) - const files = inventory.supportedFiles - const inventoryMetrics = sourceInventoryMetrics(files) - const currentFiles = new Map(files.map((file) => [file.relativePath, file])) - const policyRebuild = - !options.rebuild && - existingManifest !== null && - (existingManifest.schemaVersion !== INDEX_SCHEMA_VERSION || - existingManifest.indexPolicyFingerprint !== policyFingerprint) - const canReuse = manifestCompatible && existingTable !== null - const previousIndexedFiles = canReuse ? (existingManifest.indexedFiles ?? []) : [] - const previousEmptyFiles = canReuse ? storedEmptyFiles : [] - const reusableIndexedFiles = previousIndexedFiles.filter( - (file) => currentFiles.get(file.relativePath)?.checksum === file.checksum, - ) - const reusableEmptyFiles = previousEmptyFiles.filter( - (file) => currentFiles.get(file.relativePath)?.checksum === file.checksum, - ) - const reusableFiles = new Set([ - ...reusableIndexedFiles.map((file) => file.relativePath), - ...reusableEmptyFiles.map((file) => file.relativePath), - ]) - const filesToIndex = files.filter((file) => !reusableFiles.has(file.relativePath)) - const allChunks: TextChunk[] = [] - const errors: IngestResult["errors"] = [] - const redactionCounts: RedactionCount[] = [] - const emptyTextFiles: string[] = [] - - const results = await mapLimit(filesToIndex, config.ingestConcurrency, signal, async (file) => { - try { - const parsed = await parseFile(file, config) + let state: IngestionRunState | null = null + try { + throwIfAborted(signal) + const requestedBatchSize = ingestFileBatchSize(options.batchSize) + const policyFingerprint = indexPolicyFingerprint(config) + const existingManifest = await readIndexManifest(config) + const storedState = await readIngestionState(config) + const storedEmptyFiles = await readEmptyTextFiles(config) + const knownFiles = new Map( + [ + ...(existingManifest?.indexedFiles ?? []), + ...storedEmptyFiles, + ...(storedState?.files ?? []), + ].map((file) => [file.relativePath, file]), + ) + const inventory = await inventorySourceFiles(config, { knownFiles }) + const files = inventory.supportedFiles + const inventoryMetrics = sourceInventoryMetrics(files) + const currentFiles = new Map(files.map((file) => [file.relativePath, file])) + const existingTable = await openRowsTable(config, connection) + const manifestCompatible = + !options.rebuild && + existingManifest?.schemaVersion === INDEX_SCHEMA_VERSION && + existingManifest.indexPolicyFingerprint === policyFingerprint && + existingManifest.indexedFiles !== undefined + const policyRebuild = + !options.rebuild && + existingManifest !== null && + (existingManifest.schemaVersion !== INDEX_SCHEMA_VERSION || + existingManifest.indexPolicyFingerprint !== policyFingerprint) + const canReuse = manifestCompatible && existingTable !== null + const previousIndexedFiles = canReuse ? (existingManifest.indexedFiles ?? []) : [] + const previousEmptyFiles = canReuse ? storedEmptyFiles : [] + const reusableIndexedFiles = previousIndexedFiles.filter( + (file) => currentFiles.get(file.relativePath)?.checksum === file.checksum, + ) + const reusableEmptyFiles = previousEmptyFiles.filter( + (file) => currentFiles.get(file.relativePath)?.checksum === file.checksum, + ) + const reusablePaths = new Set([ + ...reusableIndexedFiles.map((file) => file.relativePath), + ...reusableEmptyFiles.map((file) => file.relativePath), + ]) + const activeTableName = await activeIndexTableName(config) + const resumableTable = storedState + ? await openRowsTableByName(storedState.tableName, config, connection) + : null + const resumableTableAvailable = + !storedState?.files.some((file) => file.state === "indexed" && file.chunkCount > 0) || + resumableTable !== null + + if ( + storedState && + resumableTableAvailable && + canResumeIngestion(storedState, files, policyFingerprint, options.rebuild === true) + ) { + state = resumeIngestionState(await reconcileCommittedFiles(storedState, config, connection)) + } else { + if ( + storedState?.mode === "rebuild" && + storedState.tableName !== activeTableName && + storedState.status !== "completed" && + storedState.status !== "completed_with_errors" + ) { + await dropRowsTable(storedState.tableName, config, connection) + await removeStagedIndexManifest(storedState.runId, config) + } + const mode = + options.rebuild || (existingTable !== null && !manifestCompatible) + ? "rebuild" + : "incremental" + const reusableChunkCounts = new Map([ + ...reusableIndexedFiles.map((file) => [file.relativePath, file.chunkCount] as const), + ...reusableEmptyFiles.map((file) => [file.relativePath, 0] as const), + ]) + state = createIngestionRunState({ + mode, + tableName: activeTableName, + previousTableName: mode === "rebuild" ? activeTableName : null, + policyFingerprint, + batchSize: requestedBatchSize, + files, + reusablePaths: mode === "incremental" ? reusablePaths : new Set(), + reusableChunkCounts, + }) + if (mode === "rebuild") { + state = { + ...state, + tableName: generationTableName(config.tableName, state.runId), + files: state.files.map((file) => ({ + ...file, + state: "pending", + chunkCount: 0, + reused: false, + })), + } + } + } + + await persistIngestionProgress(state, config, options) + throwIfAborted(signal) + + const previousPaths = new Set(previousIndexedFiles.map((file) => file.relativePath)) + const currentPaths = new Set(files.map((file) => file.relativePath)) + const removedPaths = + state.mode === "incremental" + ? [...previousPaths].filter((relativePath) => !currentPaths.has(relativePath)) + : [] + const pendingFiles = state.files + .filter((file) => file.state !== "indexed") + .flatMap((file) => { + const source = currentFiles.get(file.relativePath) + return source ? [source] : [] + }) + let removalApplied = false + let lexicalIndexWarning: string | null = null + + for (const fileBatch of valueBatches(pendingFiles, state.batchSize)) { throwIfAborted(signal) - const redacted = redactText(parsed.text, config) - const chunks = chunkDocument( - { ...parsed, text: redacted.text }, - config.chunkSize, - config.chunkOverlap, + const parsedBatch = await mapLimit( + fileBatch, + config.ingestConcurrency, + signal, + async (file) => parseSourceFile(file, config, signal), ) - return { path: file.relativePath, chunks, redactions: redacted.counts, error: null } - } catch (error) { + for (const parsed of parsedBatch) { + state = updateIngestionFile( + state, + parsed.file.relativePath, + parsed.error + ? { state: "error", chunkCount: 0, redactions: 0, error: parsed.error } + : { + state: "parsed", + chunkCount: parsed.chunks.length, + redactions: parsed.redactions, + error: null, + }, + ) + } + await persistIngestionProgress(state, config, options) + throwIfAborted(signal) + + const successfulFiles = parsedBatch.filter((parsed) => parsed.error === null) + const allChunks = successfulFiles.flatMap((parsed) => parsed.chunks) + const rows = await vectorRowsForChunks(allChunks, config, signal) + for (const parsed of successfulFiles) { + state = updateIngestionFile(state, parsed.file.relativePath, { state: "embedded" }) + } + await persistIngestionProgress(state, config, options) throwIfAborted(signal) - return { - path: file.relativePath, - chunks: [], - redactions: [], - error: { - path: file.relativePath, - message: error instanceof Error ? error.message : String(error), - }, + + const replacePaths = [ + ...fileBatch.map((file) => file.relativePath), + ...(!removalApplied ? removedPaths : []), + ] + const writeResult = await updateRowsInTable( + rows, + replacePaths, + state.tableName, + config, + connection, + ) + removalApplied = true + lexicalIndexWarning ??= writeResult.lexicalIndexWarning + for (const parsed of successfulFiles) { + state = updateIngestionFile(state, parsed.file.relativePath, { state: "indexed" }) } + await writeProgressManifest(state, config, connection) + await persistIngestionProgress(state, config, options) + throwIfAborted(signal) } - }) - for (const result of results) { - if (result.error) { - errors.push(result.error) - continue + if (!removalApplied && removedPaths.length > 0) { + const writeResult = await updateRowsInTable( + [], + removedPaths, + state.tableName, + config, + connection, + ) + lexicalIndexWarning ??= writeResult.lexicalIndexWarning + } + + const manifest = await manifestForState(state, config, connection) + await validateIngestionTable(state, manifest, config, connection) + await writeEmptyTextFiles(emptyTextRecords(state), config) + await writeIndexManifest(manifest, config) + await removeStagedIndexManifest(state.runId, config) + const errors = ingestionErrors(state) + state = finishIngestionState(state, errors.length > 0 ? "completed_with_errors" : "completed") + await persistIngestionProgress(state, config, options) + + const indexedFiles = manifest.indexedFiles ?? [] + const emptyTextFiles = emptyTextRecords(state).map((file) => file.relativePath) + const redactions = state.files.reduce((sum, file) => sum + file.redactions, 0) + await recordAccess(config, { + action: "ingest", + resultCount: manifest.chunkCount, + redactions, + }) + + return { + runId: state.runId, + resumed: state.resumed, + batchSize: state.batchSize, + indexedFiles: indexedFiles.length, + rebuiltFiles: state.files.filter( + (file) => file.state === "indexed" && file.chunkCount > 0 && !file.reused, + ).length, + reusedFiles: state.files.filter((file) => file.reused).length, + policyRebuild, + chunks: manifest.chunkCount, + discoveredFiles: inventory.discoveredFiles, + supportedFiles: files.length, + supportedBytes: inventoryMetrics.supportedBytes, + largestFileBytes: inventoryMetrics.largestFileBytes, + skippedFiles: inventory.skippedFiles.length + emptyTextFiles.length, + unsupportedFiles: countSkippedByReason(inventory.skippedFiles, "unsupported-extension"), + oversizedFiles: countSkippedByReason(inventory.skippedFiles, "oversized"), + sensitiveFiles: countSkippedByReason(inventory.skippedFiles, "sensitive-name"), + emptyTextFiles, + unsupportedExtensions: summarizeUnsupportedExtensions(inventory.skippedFiles), + redactions, + vectorIndexWarning: null, + lexicalIndexWarning, + errors, + } + } catch (error) { + if (state?.status === "running") { + state = finishIngestionState(state, signal?.aborted ? "interrupted" : "failed") + try { + await writeIngestionState(state, config) + } catch { + // Preserve the original ingestion failure. + } + } + throw error + } +} + +async function reconcileCommittedFiles( + state: IngestionRunState, + config: Config, + connection: Connection | undefined, +): Promise { + const filesToReconcile = state.files.filter( + (file) => file.state !== "indexed" && file.state !== "error", + ) + if (filesToReconcile.length === 0) { + return state + } + + const table = await openRowsTableByName(state.tableName, config, connection) + const rows = table + ? ((await table.query().select(["relativePath", "checksum"]).toArray()) as Array<{ + relativePath: string + checksum: string + }>) + : [] + const rowCounts = new Map() + for (const row of rows) { + const key = `${row.relativePath}\0${row.checksum}` + rowCounts.set(key, (rowCounts.get(key) ?? 0) + 1) + } + const committedEmptyFiles = + state.mode === "incremental" + ? new Set( + (await readEmptyTextFiles(config)).map( + (file) => `${file.relativePath}\0${file.checksum}`, + ), + ) + : new Set() + + return { + ...state, + files: state.files.map((file) => { + if (file.state === "indexed" || file.state === "error") { + return file + } + const key = `${file.relativePath}\0${file.checksum}` + const committed = + file.chunkCount > 0 ? rowCounts.get(key) === file.chunkCount : committedEmptyFiles.has(key) + return committed ? { ...file, state: "indexed", error: null } : file + }), + } +} + +interface ParsedSourceFile { + file: SourceFile + chunks: TextChunk[] + redactions: number + error: string | null +} + +async function parseSourceFile( + file: SourceFile, + config: Config, + signal: AbortSignal | undefined, +): Promise { + try { + const parsed = await parseFile(file, config) + throwIfAborted(signal) + const redacted = redactText(parsed.text, config) + return { + file, + chunks: chunkDocument( + { ...parsed, text: redacted.text }, + config.chunkSize, + config.chunkOverlap, + ), + redactions: totalRedactions(redacted.counts), + error: null, } - redactionCounts.push(...result.redactions) - if (result.chunks.length === 0) { - emptyTextFiles.push(result.path) + } catch (error) { + throwIfAborted(signal) + return { + file, + chunks: [], + redactions: 0, + error: error instanceof Error ? error.message : String(error), } - allChunks.push(...result.chunks) } +} +async function vectorRowsForChunks( + chunks: TextChunk[], + config: Config, + signal: AbortSignal | undefined, +): Promise { const rows: VectorRow[] = [] - for (let i = 0; i < allChunks.length; i += config.embeddingBatchSize) { + for (const batch of valueBatches(chunks, config.embeddingBatchSize)) { throwIfAborted(signal) - const batch = allChunks.slice(i, i + config.embeddingBatchSize) const embeddings = await embedTexts(batch.map(chunkSearchText), config) throwIfAborted(signal) for (const [index, chunk] of batch.entries()) { @@ -178,122 +441,161 @@ async function ingestUnlocked( }) } } + return rows +} - const rebuiltIndexedFiles = indexedFileRecords(rows) - const indexedFiles = [...reusableIndexedFiles, ...rebuiltIndexedFiles].sort((a, b) => - a.relativePath.localeCompare(b.relativePath), - ) +async function persistIngestionProgress( + state: IngestionRunState, + config: Config, + options: IngestOptions, +): Promise { + await writeIngestionState(state, config) + await options.onProgress?.(ingestionProgress(state)) +} + +async function writeProgressManifest( + state: IngestionRunState, + config: Config, + connection: Connection | undefined, +): Promise { + const manifest = await manifestForState(state, config, connection) + if (state.mode === "rebuild") { + await writeStagedIndexManifest(manifest, state.runId, config) + return + } + await writeEmptyTextFiles(emptyTextRecords(state), config) + await writeIndexManifest(manifest, config) +} + +async function manifestForState( + state: IngestionRunState, + config: Config, + connection: Connection | undefined, +): Promise { + const indexedFiles = indexedFilesFromState(state) const chunkCount = indexedFiles.reduce((sum, file) => sum + file.chunkCount, 0) - const previousPaths = new Set(previousIndexedFiles.map((file) => file.relativePath)) - const currentPaths = new Set(files.map((file) => file.relativePath)) - const replacePaths = [ - ...filesToIndex.map((file) => file.relativePath), - ...[...previousPaths].filter((relativePath) => !currentPaths.has(relativePath)), - ] - throwIfAborted(signal) - const writeResult = - !canReuse || chunkCount === 0 - ? await writeRows(rows, config, connection) - : replacePaths.length > 0 - ? await updateRows(rows, replacePaths, config, connection) - : { vectorIndexWarning: null, lexicalIndexWarning: null } - if (chunkCount > 0) { - const firstRow = rows[0] ?? (await firstStoredRow(config, connection)) - if (!firstRow) { - throw new Error("Cannot write an index manifest without indexed rows.") + const table = await openRowsTableByName(state.tableName, config, connection) + const [firstRow] = table ? ((await table.query().limit(1).toArray()) as VectorRow[]) : [] + if (chunkCount > 0 && !firstRow) { + throw new Error("Cannot write an index manifest without indexed rows.") + } + return { + schemaVersion: INDEX_SCHEMA_VERSION, + createdAt: new Date().toISOString(), + ragmirVersion: VERSION, + embeddingProvider: config.embeddingProvider, + embeddingModel: config.embeddingModel, + indexPolicyFingerprint: state.policyFingerprint, + ...(firstRow ? { vectorDimension: firstRow.vector.length } : {}), + vectorDistanceMetric: VECTOR_DISTANCE_METRIC, + chunkSize: config.chunkSize, + chunkOverlap: config.chunkOverlap, + fileCount: indexedFiles.length, + chunkCount, + tableName: state.tableName, + indexedFiles, + } +} + +function indexedFilesFromState(state: IngestionRunState): IndexManifestFile[] { + return state.files + .filter((file) => file.state === "indexed" && file.chunkCount > 0) + .map((file) => ({ + relativePath: file.relativePath, + checksum: file.checksum, + chunkCount: file.chunkCount, + bytes: file.bytes, + mtimeMs: file.mtimeMs, + })) + .sort((left, right) => left.relativePath.localeCompare(right.relativePath)) +} + +function emptyTextRecords(state: IngestionRunState): Array<{ + relativePath: string + checksum: string + bytes: number + mtimeMs: number +}> { + return state.files + .filter((file) => file.state === "indexed" && file.chunkCount === 0) + .map((file) => ({ + relativePath: file.relativePath, + checksum: file.checksum, + bytes: file.bytes, + mtimeMs: file.mtimeMs, + })) + .sort((left, right) => left.relativePath.localeCompare(right.relativePath)) +} + +function ingestionErrors(state: IngestionRunState): IngestResult["errors"] { + return state.files.flatMap((file) => + file.state === "error" && file.error ? [{ path: file.relativePath, message: file.error }] : [], + ) +} + +async function validateIngestionTable( + state: IngestionRunState, + manifest: IndexManifest, + config: Config, + connection: Connection | undefined, +): Promise { + const table = await openRowsTableByName(state.tableName, config, connection) + if (!table) { + if (manifest.chunkCount === 0) { + return } - await writeIndexManifest( - { - schemaVersion: INDEX_SCHEMA_VERSION, - createdAt: new Date().toISOString(), - ragmirVersion: VERSION, - embeddingProvider: config.embeddingProvider, - embeddingModel: config.embeddingModel, - indexPolicyFingerprint: policyFingerprint, - vectorDimension: firstRow.vector.length, - vectorDistanceMetric: VECTOR_DISTANCE_METRIC, - chunkSize: config.chunkSize, - chunkOverlap: config.chunkOverlap, - fileCount: indexedFiles.length, - chunkCount, - indexedFiles, - }, - config, + throw new Error("Ingestion validation failed because the generated table is missing.") + } + const rows = (await table.query().select(["id", "relativePath", "checksum"]).toArray()) as Array<{ + id: string + relativePath: string + checksum: string + }> + if (rows.length !== manifest.chunkCount) { + throw new Error( + `Ingestion validation expected ${manifest.chunkCount} rows but found ${rows.length}.`, ) } - await writeEmptyTextFiles( - [ - ...reusableEmptyFiles, - ...emptyTextFiles.flatMap((relativePath) => { - const file = currentFiles.get(relativePath) - return file - ? [ - { - relativePath, - checksum: file.checksum, - bytes: file.bytes, - mtimeMs: file.mtimeMs, - }, - ] - : [] - }), - ], - config, + if (new Set(rows.map((row) => row.id)).size !== rows.length) { + throw new Error("Ingestion validation found duplicate chunk identifiers.") + } + const expectedFiles = new Map( + (manifest.indexedFiles ?? []).map((file) => [ + `${file.relativePath}\0${file.checksum}`, + file.chunkCount, + ]), ) - await recordAccess(config, { - action: "ingest", - resultCount: chunkCount, - redactions: totalRedactions(redactionCounts), - }) - - return { - indexedFiles: indexedFiles.length, - rebuiltFiles: new Set(rows.map((row) => row.relativePath)).size, - reusedFiles: reusableFiles.size, - policyRebuild, - chunks: chunkCount, - discoveredFiles: inventory.discoveredFiles, - supportedFiles: files.length, - supportedBytes: inventoryMetrics.supportedBytes, - largestFileBytes: inventoryMetrics.largestFileBytes, - skippedFiles: inventory.skippedFiles.length + emptyTextFiles.length, - unsupportedFiles: countSkippedByReason(inventory.skippedFiles, "unsupported-extension"), - oversizedFiles: countSkippedByReason(inventory.skippedFiles, "oversized"), - sensitiveFiles: countSkippedByReason(inventory.skippedFiles, "sensitive-name"), - emptyTextFiles: [ - ...reusableEmptyFiles.map((file) => file.relativePath), - ...emptyTextFiles, - ].sort(), - unsupportedExtensions: summarizeUnsupportedExtensions(inventory.skippedFiles), - redactions: totalRedactions(redactionCounts), - vectorIndexWarning: writeResult.vectorIndexWarning, - lexicalIndexWarning: writeResult.lexicalIndexWarning, - errors, + const actualFiles = new Map() + for (const row of rows) { + const key = `${row.relativePath}\0${row.checksum}` + actualFiles.set(key, (actualFiles.get(key) ?? 0) + 1) + } + if ( + expectedFiles.size !== actualFiles.size || + [...expectedFiles].some(([key, count]) => actualFiles.get(key) !== count) + ) { + throw new Error("Ingestion validation found rows that do not match the generated manifest.") } } -function indexedFileRecords(rows: VectorRow[]): IndexManifestFile[] { - const records = new Map() - for (const row of rows) { - const current = records.get(row.relativePath) - records.set(row.relativePath, { - relativePath: row.relativePath, - checksum: row.checksum, - chunkCount: (current?.chunkCount ?? 0) + 1, - bytes: row.bytes, - mtimeMs: row.mtimeMs, - }) +function generationTableName(baseName: string, runId: string): string { + return `${baseName}__generation_${runId.replaceAll("-", "")}` +} + +function ingestFileBatchSize(value: number | undefined): number { + const batchSize = value ?? DEFAULT_INGEST_FILE_BATCH_SIZE + if (!Number.isInteger(batchSize) || batchSize <= 0) { + throw new Error("batchSize must be a positive integer.") } - return [...records.values()] + return batchSize } -async function firstStoredRow(config: Config, connection?: Connection): Promise { - const table = await openRowsTable(config, connection) - if (!table) { - return null +function valueBatches(values: T[], batchSize: number): T[][] { + const batches: T[][] = [] + for (let index = 0; index < values.length; index += batchSize) { + batches.push(values.slice(index, index + batchSize)) } - const [row] = (await table.query().limit(1).toArray()) as VectorRow[] - return row ?? null + return batches } export async function audit(cwd = process.cwd()): Promise { diff --git a/packages/ragmir-core/src/ingestion-state.ts b/packages/ragmir-core/src/ingestion-state.ts new file mode 100644 index 0000000..d5eeb63 --- /dev/null +++ b/packages/ragmir-core/src/ingestion-state.ts @@ -0,0 +1,305 @@ +import { randomUUID } from "node:crypto" +import { readFile, rename, rm, writeFile } from "node:fs/promises" +import path from "node:path" +import { isRecord } from "./guards.js" +import { ensurePrivateDirectory, hardenPrivateFile } from "./permissions.js" +import type { + Config, + IndexManifest, + IngestionFileStage, + IngestionProgress, + IngestionRunMode, + IngestionRunStatus, + SourceFile, +} from "./types.js" + +const INGESTION_STATE_VERSION = 1 +const INGESTION_STATE_FILENAME = "ingestion-state.json" + +export interface IngestionFileState { + relativePath: string + checksum: string + bytes: number + mtimeMs: number + policyFingerprint: string + state: IngestionFileStage + chunkCount: number + redactions: number + error: string | null + reused: boolean + updatedAt: string +} + +export interface IngestionRunState { + version: number + runId: string + mode: IngestionRunMode + status: IngestionRunStatus + tableName: string + previousTableName: string | null + policyFingerprint: string + batchSize: number + createdAt: string + updatedAt: string + lastActivityAt: string + resumed: boolean + files: IngestionFileState[] +} + +interface CreateIngestionRunStateOptions { + mode: IngestionRunMode + tableName: string + previousTableName: string | null + policyFingerprint: string + batchSize: number + files: SourceFile[] + reusablePaths: Set + reusableChunkCounts: Map +} + +export function createIngestionRunState( + options: CreateIngestionRunStateOptions, +): IngestionRunState { + const now = new Date().toISOString() + return { + version: INGESTION_STATE_VERSION, + runId: randomUUID(), + mode: options.mode, + status: "running", + tableName: options.tableName, + previousTableName: options.previousTableName, + policyFingerprint: options.policyFingerprint, + batchSize: options.batchSize, + createdAt: now, + updatedAt: now, + lastActivityAt: now, + resumed: false, + files: options.files.map((file) => { + const reused = options.reusablePaths.has(file.relativePath) + return { + relativePath: file.relativePath, + checksum: file.checksum, + bytes: file.bytes, + mtimeMs: file.mtimeMs, + policyFingerprint: options.policyFingerprint, + state: reused ? "indexed" : "pending", + chunkCount: options.reusableChunkCounts.get(file.relativePath) ?? 0, + redactions: 0, + error: null, + reused, + updatedAt: now, + } + }), + } +} + +export function canResumeIngestion( + state: IngestionRunState, + files: SourceFile[], + policyFingerprint: string, + rebuildRequested: boolean, +): boolean { + if ( + !["running", "interrupted", "failed"].includes(state.status) || + state.policyFingerprint !== policyFingerprint || + (rebuildRequested && state.mode !== "rebuild") || + state.files.length !== files.length + ) { + return false + } + + const currentFiles = new Map(files.map((file) => [file.relativePath, file])) + return state.files.every((file) => { + const current = currentFiles.get(file.relativePath) + return current?.checksum === file.checksum && current.bytes === file.bytes + }) +} + +export function resumeIngestionState(state: IngestionRunState): IngestionRunState { + const now = new Date().toISOString() + return { + ...state, + status: "running", + resumed: true, + updatedAt: now, + lastActivityAt: now, + files: state.files.map((file) => + file.state === "indexed" + ? file + : { ...file, state: "pending", chunkCount: 0, error: null, updatedAt: now }, + ), + } +} + +export function updateIngestionFile( + state: IngestionRunState, + relativePath: string, + update: Partial< + Pick + >, +): IngestionRunState { + const now = new Date().toISOString() + return { + ...state, + updatedAt: now, + lastActivityAt: now, + files: state.files.map((file) => + file.relativePath === relativePath ? { ...file, ...update, updatedAt: now } : file, + ), + } +} + +export function finishIngestionState( + state: IngestionRunState, + status: Exclude, +): IngestionRunState { + const now = new Date().toISOString() + return { + ...state, + status, + updatedAt: now, + lastActivityAt: now, + } +} + +export async function writeIngestionState(state: IngestionRunState, config: Config): Promise { + await writePrivateJsonAtomic( + path.join(config.storageDir, INGESTION_STATE_FILENAME), + state, + config.storageDir, + ) +} + +export async function writeStagedIndexManifest( + manifest: IndexManifest, + runId: string, + config: Config, +): Promise { + await writePrivateJsonAtomic(stagedManifestPath(runId, config), manifest, config.storageDir) +} + +export async function removeStagedIndexManifest(runId: string, config: Config): Promise { + await rm(stagedManifestPath(runId, config), { force: true }) +} + +export async function readIngestionState(config: Config): Promise { + try { + const value = JSON.parse( + await readFile(path.join(config.storageDir, INGESTION_STATE_FILENAME), "utf8"), + ) as unknown + return isIngestionRunState(value) ? value : null + } catch (error) { + if (error instanceof SyntaxError || (isNodeError(error) && error.code === "ENOENT")) { + return null + } + throw error + } +} + +export async function getIngestionProgress(config: Config): Promise { + const state = await readIngestionState(config) + return state ? ingestionProgress(state) : null +} + +export function ingestionProgress(state: IngestionRunState): IngestionProgress { + const count = (fileState: IngestionFileStage): number => + state.files.filter((file) => file.state === fileState).length + return { + runId: state.runId, + mode: state.mode, + status: state.status, + resumed: state.resumed, + batchSize: state.batchSize, + totalFiles: state.files.length, + pendingFiles: count("pending"), + parsedFiles: count("parsed"), + embeddedFiles: count("embedded"), + indexedFiles: count("indexed"), + errorFiles: count("error"), + chunksIndexed: state.files + .filter((file) => file.state === "indexed") + .reduce((sum, file) => sum + file.chunkCount, 0), + lastActivityAt: state.lastActivityAt, + } +} + +function isIngestionRunState(value: unknown): value is IngestionRunState { + return ( + isRecord(value) && + value.version === INGESTION_STATE_VERSION && + typeof value.runId === "string" && + (value.mode === "incremental" || value.mode === "rebuild") && + isIngestionRunStatus(value.status) && + typeof value.tableName === "string" && + (value.previousTableName === null || typeof value.previousTableName === "string") && + typeof value.policyFingerprint === "string" && + typeof value.batchSize === "number" && + typeof value.createdAt === "string" && + typeof value.updatedAt === "string" && + typeof value.lastActivityAt === "string" && + typeof value.resumed === "boolean" && + Array.isArray(value.files) && + value.files.every(isIngestionFileState) + ) +} + +function isIngestionRunStatus(value: unknown): value is IngestionRunStatus { + return ( + value === "running" || + value === "interrupted" || + value === "failed" || + value === "completed" || + value === "completed_with_errors" + ) +} + +function stagedManifestPath(runId: string, config: Config): string { + return path.join(config.storageDir, `index-manifest.${runId}.staging.json`) +} + +function isIngestionFileState(value: unknown): value is IngestionFileState { + return ( + isRecord(value) && + typeof value.relativePath === "string" && + typeof value.checksum === "string" && + typeof value.bytes === "number" && + typeof value.mtimeMs === "number" && + typeof value.policyFingerprint === "string" && + isIngestionFileStage(value.state) && + typeof value.chunkCount === "number" && + typeof value.redactions === "number" && + (value.error === null || typeof value.error === "string") && + typeof value.reused === "boolean" && + typeof value.updatedAt === "string" + ) +} + +function isIngestionFileStage(value: unknown): value is IngestionFileStage { + return ( + value === "pending" || + value === "parsed" || + value === "embedded" || + value === "indexed" || + value === "error" + ) +} + +function isNodeError(error: unknown): error is NodeJS.ErrnoException { + return error instanceof Error && "code" in error +} + +async function writePrivateJsonAtomic( + targetPath: string, + value: unknown, + directory: string, +): Promise { + await ensurePrivateDirectory(directory) + const temporaryPath = `${targetPath}.${process.pid}.${randomUUID()}.tmp` + try { + await writeFile(temporaryPath, JSON.stringify(value, null, 2), "utf8") + await hardenPrivateFile(temporaryPath) + await rename(temporaryPath, targetPath) + } finally { + await rm(temporaryPath, { force: true }) + } +} diff --git a/packages/ragmir-core/src/store.test.ts b/packages/ragmir-core/src/store.test.ts index 4f926f7..a25bf75 100644 --- a/packages/ragmir-core/src/store.test.ts +++ b/packages/ragmir-core/src/store.test.ts @@ -10,6 +10,7 @@ import { writeEmptyTextFiles, writeIndexManifest, writeRows, + writeRowsToTable, } from "./store.js" import { testConfig } from "./test-support/config.js" import type { IndexManifest } from "./types.js" @@ -109,6 +110,39 @@ describe("store", () => { expect(result.vectorIndexWarning).toBeNull() expect(result.lexicalIndexWarning).toBeNull() }) + + it("activates a completed generation through the manifest while preserving legacy tables", async () => { + const root = await mkdtemp(path.join(os.tmpdir(), "ragmir-store-generation-")) + tempDirs.push(root) + const config = testConfig(root) + const generationTable = "chunks__generation_test" + await writeRows([sampleRow(".ragmir/raw/legacy.md", 0, [0.1, 0.2], config)], config) + await writeRowsToTable( + [sampleRow(".ragmir/raw/current.md", 0, [0.3, 0.4], config)], + generationTable, + config, + ) + + await writeIndexManifest( + { + schemaVersion: 7, + createdAt: "2026-07-14T00:00:00.000Z", + ragmirVersion: "test", + embeddingProvider: config.embeddingProvider, + embeddingModel: config.embeddingModel, + chunkSize: config.chunkSize, + chunkOverlap: config.chunkOverlap, + fileCount: 1, + chunkCount: 1, + tableName: generationTable, + }, + config, + ) + + expect((await readRows(config)).map((row) => row.relativePath)).toEqual([ + ".ragmir/raw/current.md", + ]) + }) }) describe("empty-text-files manifest", () => { diff --git a/packages/ragmir-core/src/store.ts b/packages/ragmir-core/src/store.ts index 5d4f828..5078c6c 100644 --- a/packages/ragmir-core/src/store.ts +++ b/packages/ragmir-core/src/store.ts @@ -32,19 +32,32 @@ export async function writeRows( rows: VectorRow[], config: Config, connection?: lancedb.Connection, +): Promise { + const tableName = await activeIndexTableName(config) + const result = await writeRowsToTable(rows, tableName, config, connection) + if (rows.length === 0) { + await rm(path.join(config.storageDir, INDEX_MANIFEST_FILENAME), { force: true }) + } + return result +} + +export async function writeRowsToTable( + rows: VectorRow[], + tableName: string, + config: Config, + connection?: lancedb.Connection, ): Promise { return withConnection(config, connection, async (db) => { if (rows.length === 0) { const tableNames = await db.tableNames() - if (tableNames.includes(config.tableName)) { - await db.dropTable(config.tableName) + if (tableNames.includes(tableName)) { + await db.dropTable(tableName) } - await rm(path.join(config.storageDir, INDEX_MANIFEST_FILENAME), { force: true }) return { vectorIndexWarning: null, lexicalIndexWarning: null } } const records = storedRows(rows) - const table = await db.createTable(config.tableName, records, { + const table = await db.createTable(tableName, records, { mode: "overwrite", }) @@ -62,9 +75,20 @@ export async function updateRows( config: Config, connection?: lancedb.Connection, ): Promise { - const table = await openRowsTable(config, connection) + const tableName = await activeIndexTableName(config) + return updateRowsInTable(rows, replacePaths, tableName, config, connection) +} + +export async function updateRowsInTable( + rows: VectorRow[], + replacePaths: string[], + tableName: string, + config: Config, + connection?: lancedb.Connection, +): Promise { + const table = await openRowsTableByName(tableName, config, connection) if (!table) { - return writeRows(rows, config, connection) + return writeRowsToTable(rows, tableName, config, connection) } for (const paths of batches([...new Set(replacePaths)], 200)) { @@ -151,6 +175,7 @@ function isIndexManifest(value: unknown): value is IndexManifest { typeof value.chunkOverlap === "number" && typeof value.fileCount === "number" && typeof value.chunkCount === "number" && + (!("tableName" in value) || typeof value.tableName === "string") && (!("indexedFiles" in value) || (Array.isArray(value.indexedFiles) && value.indexedFiles.every(isIndexManifestFile))) ) @@ -205,13 +230,37 @@ export async function readEmptyTextFiles(config: Config): Promise { + return openRowsTableByName(await activeIndexTableName(config), config, connection) +} + +export async function openRowsTableByName( + tableName: string, + config: Config, + connection?: lancedb.Connection, ): Promise { return withConnection(config, connection, async (db) => { const tableNames = await db.tableNames() - if (!tableNames.includes(config.tableName)) { + if (!tableNames.includes(tableName)) { return null } - return db.openTable(config.tableName) + return db.openTable(tableName) + }) +} + +export async function activeIndexTableName(config: Config): Promise { + return (await readIndexManifest(config))?.tableName ?? config.tableName +} + +export async function dropRowsTable( + tableName: string, + config: Config, + connection?: lancedb.Connection, +): Promise { + await withConnection(config, connection, async (db) => { + if ((await db.tableNames()).includes(tableName)) { + await db.dropTable(tableName) + } }) } diff --git a/packages/ragmir-core/src/types.ts b/packages/ragmir-core/src/types.ts index f9f8ad3..98becac 100644 --- a/packages/ragmir-core/src/types.ts +++ b/packages/ragmir-core/src/types.ts @@ -114,6 +114,7 @@ export interface IndexManifest { chunkOverlap: number fileCount: number chunkCount: number + tableName?: string indexedFiles?: IndexManifestFile[] } @@ -228,9 +229,14 @@ export interface OperationOptions { export interface IngestOptions extends OperationOptions { cwd?: PathLike rebuild?: boolean + batchSize?: number + onProgress?: (progress: IngestionProgress) => void | Promise } export interface IngestResult { + runId: string + resumed: boolean + batchSize: number discoveredFiles: number supportedFiles: number supportedBytes: number @@ -252,6 +258,33 @@ export interface IngestResult { errors: Array<{ path: string; message: string }> } +export type IngestionFileStage = "pending" | "parsed" | "embedded" | "indexed" | "error" + +export type IngestionRunMode = "incremental" | "rebuild" + +export type IngestionRunStatus = + | "running" + | "interrupted" + | "failed" + | "completed" + | "completed_with_errors" + +export interface IngestionProgress { + runId: string + mode: IngestionRunMode + status: IngestionRunStatus + resumed: boolean + batchSize: number + totalFiles: number + pendingFiles: number + parsedFiles: number + embeddedFiles: number + indexedFiles: number + errorFiles: number + chunksIndexed: number + lastActivityAt: string +} + export interface PreviewChunksOptions { cwd?: PathLike paths?: string[]