diff --git a/knowledge-fs/.harness/changes/2026-07-30-document-compilation-retry-recovery.md b/knowledge-fs/.harness/changes/2026-07-30-document-compilation-retry-recovery.md new file mode 100644 index 00000000000..fe2c901ed68 --- /dev/null +++ b/knowledge-fs/.harness/changes/2026-07-30-document-compilation-retry-recovery.md @@ -0,0 +1,83 @@ +# Document compilation retry recovery + +Date: 2026-07-30 + +## What changed + +- Parse-artifact upserts now keep parser-generated element ids bound to the first persisted + artifact id for a document version. Database-backed retries also repair rows written by the old + behavior, while custom/non-generated element ids remain unchanged. +- Generation-scoped document compilation now resumes from durable checkpoints: + - `parsed` and later checkpoints reload the canonical parse artifact instead of parsing the + source object again. + - `outline_built` and later checkpoints reload and validate the persisted outline and multimodal + manifest instead of rebuilding the outline, rerunning its LLM summaries, or rematerializing + PageIndex. + - Candidate receipt path ids are reconstructed deterministically from the persisted generation. +- An `outline_built` retry removes only failed projections belonging to the unpublished + generation's deterministic node ids before rebuilding FTS and embeddings. Missing or + inconsistent checkpoint data fails closed. +- Added regression coverage for canonical element-id replay, checkpoint resume, incomplete or + inconsistent checkpoint state, embedding-timeout projection recovery, and cleanup preconditions. +- No frontend, API contract, schema, or migration files changed. + +## Why + +After an embedding/model-runtime timeout, the old retry path parsed the same file again with a new +random parse-artifact id. The logical parse-artifact row retained its original outer id, but its +elements were overwritten with ids derived from the new random id. Rebuilding the already-persisted +generation outline therefore changed an immutable logical value and failed with +`Generation-scoped document-outline ... conflicts with its immutable persisted value`. + +The retry also repeated expensive parser, outline-summary LLM, and PageIndex work that had already +completed. Resuming from the durable checkpoint preserves immutable generation identity and limits +the retry to the failed indexing work. + +## Database access and performance + +- Each compilation execution adds one bounded attempt lookup. A resumed execution performs one + parse-artifact lookup and canonical upsert, with one conditional repair update only for legacy + inconsistent generated ids. +- Outline and multimodal-manifest checkpoint reads run in parallel and use the existing + `(document_asset_id, version, publication_generation)` unique access paths. +- Failed projection cleanup is batched by the configured projection batch size, never queried or + deleted per node, and is bounded to at most three projection types per node. Existing + node/projection indexes cover these ids; no new index is required. +- The existing `parse_artifacts_asset_version_uq`, + `document_outlines_asset_version_uq`, + `document_multimodal_manifests_asset_version_uq`, and + `index_projections_node_type_version_idx` indexes remain the required access paths. + +## Verification + +- TDD red phase reproduced all three failure boundaries: + - retry-generated element ids did not match the persisted parse-artifact id; + - a failed FTS projection conflicted during embedding retry; + - an `outline_built` retry reparsed the document instead of resuming. +- Focused regression suite: + - `pnpm exec vitest run src/parse-artifact-repository.test.ts src/index-reindexer.test.ts src/document-compilation-worker.test.ts` + - 3 files, 30 tests passed. +- Full API suite: + - 373 files passed, 4,104 tests passed, 3 existing environment-dependent tests skipped. +- API coverage: + - 93.88% lines/statements, 96.34% functions, 90.01% branches. +- `pnpm build` passed for all 12 KnowledgeFS packages. +- `pnpm check` passed, including workspace typechecks/tests, contract determinism, non-API coverage, + evaluations, migration checks, workflow checks, Compose validation, and static Docker smoke + checks. +- Biome passed for all six changed TypeScript files. +- Repository-wide `pnpm lint` remains blocked by pre-existing formatting findings in unchanged + Admin/test/generated-contract files and the existing 1.3 MiB OpenAPI artifact exceeding Biome's + 1 MiB processing limit. + +## Risks and follow-up + +- Cleanup deliberately deletes only `failed` projections in an unpublished generation. If a + process is killed before building projections are marked failed, retry still fails closed rather + than deleting ambiguous state; the existing candidate GC/operator recovery path remains required + for that uncommon crash boundary. +- Automatic repair recognizes only the parser-owned + `:element-` convention. This prevents the recovery path from silently + rewriting connector- or operator-owned custom element ids. +- Existing affected jobs need only be retried after deploying this code; no data migration or + document re-upload is required. diff --git a/knowledge-fs/packages/api/src/document-compilation-worker.test.ts b/knowledge-fs/packages/api/src/document-compilation-worker.test.ts index e79496abbdf..40a99a69a6d 100644 --- a/knowledge-fs/packages/api/src/document-compilation-worker.test.ts +++ b/knowledge-fs/packages/api/src/document-compilation-worker.test.ts @@ -12,6 +12,7 @@ import { import { createDocumentCompilationJobStateMachine, createDocumentCompilationWorker, + createDocumentMultimodalManifestBuilder, createDocumentOutlineBuilder, createDocumentOutlineSummaryEnhancer, createInMemoryDocumentAssetRepository, @@ -591,6 +592,226 @@ describe("createDocumentCompilationWorker lease integration", () => { ).resolves.toMatchObject({ parserStatus: "pending" }); }); + it("resumes an outline-built generation without reparsing or regenerating LLM summaries", async () => { + const adapter = createNodePlatformAdapter({ env: {} }); + const assets = createInMemoryDocumentAssetRepository({ + maxAssets: 1, + now: () => "2026-07-30T15:16:20.000Z", + }); + const asset = await assets.create({ + filename: "Retry.md", + id: "018f0d60-7a49-7cc2-9c1b-5b36f18f7001", + knowledgeSpaceId: "018f0d60-7a49-7cc2-9c1b-5b36f18f2c42", + mimeType: "text/markdown", + objectKey: "tenant-1/spaces/space/documents/asset/Retry.md", + sha256: "a".repeat(64), + sizeBytes: 7, + }); + await adapter.objectStorage.putObject({ + body: new TextEncoder().encode("# Retry"), + contentType: asset.mimeType, + key: asset.objectKey, + metadata: {}, + }); + const generationId = "018f0d60-7a49-7cc2-9c1b-5b36f18f7002"; + const canonicalArtifact = ParseArtifactSchema.parse({ + artifactHash: "b".repeat(64), + contentType: "text", + createdAt: "2026-07-30T15:16:21.000Z", + documentAssetId: asset.id, + elements: [ + { + id: "018f0d60-7a49-7cc2-9c1b-5b36f18f7003:element-1", + metadata: {}, + sectionPath: ["Retry"], + text: "Retry content", + type: "heading", + }, + ], + id: "018f0d60-7a49-7cc2-9c1b-5b36f18f7003", + metadata: {}, + parser: "native-markdown", + version: asset.version, + }); + const artifacts = createInMemoryParseArtifactRepository({ maxArtifacts: 2 }); + await artifacts.create(canonicalArtifact); + const outlines = createInMemoryDocumentOutlineRepository({ maxOutlines: 2 }); + const outlineBuilder = createDocumentOutlineBuilder({ + maxElements: 10, + maxNodes: 10, + maxSummaryChars: 200, + now: () => "2026-07-30T15:16:22.000Z", + }); + const persistedOutline = await outlines.upsert( + outlineBuilder.build({ + knowledgeSpaceId: asset.knowledgeSpaceId, + parseArtifact: canonicalArtifact, + publicationGenerationId: generationId, + }), + ); + const multimodalManifests = createInMemoryDocumentMultimodalManifestRepository({ + maxManifests: 2, + }); + const persistedManifest = await multimodalManifests.upsert( + createDocumentMultimodalManifestBuilder().build({ + artifact: canonicalArtifact, + knowledgeSpaceId: asset.knowledgeSpaceId, + publicationGenerationId: generationId, + }), + ); + const compilationJobs = createDocumentCompilationJobStateMachine({ + generateId: () => "document-compilation-job-outline-retry-1", + generatePublicationGenerationId: () => generationId, + jobs: adapter.jobs, + repository: createInMemoryDocumentCompilationJobRepository({ maxJobs: 1 }), + }); + const compilationJob = await compilationJobs.start({ + documentAssetId: asset.id, + knowledgeSpaceId: asset.knowledgeSpaceId, + tenantId: "tenant-1", + version: asset.version, + }); + await compilationJobs.advance(compilationJob.id, "parsed"); + await compilationJobs.advance(compilationJob.id, "outline_built"); + let parserCalls = 0; + let summaryCalls = 0; + let canonicalArtifactAvailable = false; + let manifestCheckpoint: "invalid" | "missing" | "valid" = "missing"; + const receipts: unknown[] = []; + const resetFailedProjectionFlags: Array = []; + const checkpointManifests = { + ...multimodalManifests, + getByDocumentVersion: async ( + input: Parameters[0], + ) => { + const manifest = await multimodalManifests.getByDocumentVersion(input); + if (manifestCheckpoint === "missing" || !manifest) { + return null; + } + + return manifestCheckpoint === "invalid" + ? { ...manifest, artifactHash: "c".repeat(64) } + : manifest; + }, + }; + const workerWithoutCheckpointLoader = createDocumentCompilationWorker({ + assets, + candidateComposer: { compose: async () => undefined }, + failureManagement: "caller", + generateKnowledgePathId: () => "018f0d60-7a49-7cc2-9c1b-5b36f18f7004", + jobs: compilationJobs, + knowledgePaths: createInMemoryKnowledgePathRepository({ + maxListLimit: 20, + maxPaths: 20, + }), + multimodalManifests, + objectStorage: adapter.objectStorage, + outlineBuilder, + outlines, + pageIndexBuild: { + materializeBuilding: async () => { + throw new Error("PageIndex must not run without a checkpoint artifact loader"); + }, + }, + parser: parser(), + reindexer: { + reindex: async () => { + throw new Error("reindex must not run without a checkpoint artifact loader"); + }, + }, + }); + const payload = { + documentAssetId: asset.id, + documentCompilationJobId: compilationJob.id, + knowledgeSpaceId: asset.knowledgeSpaceId, + publicationGenerationId: generationId, + tenantId: "tenant-1", + version: asset.version, + } as const; + + await expect( + workerWithoutCheckpointLoader.process({ + ...payload, + documentCompilationJobId: "missing-document-compilation-job", + }), + ).rejects.toThrow("Document compilation job not found"); + await expect(workerWithoutCheckpointLoader.process(payload)).rejects.toThrow( + "cannot load its parse artifact", + ); + + const worker = createDocumentCompilationWorker({ + assets, + candidateComposer: { + compose: async (input) => { + receipts.push(input); + }, + }, + failureManagement: "caller", + generateKnowledgePathId: () => "018f0d60-7a49-7cc2-9c1b-5b36f18f7004", + jobs: compilationJobs, + knowledgePaths: createInMemoryKnowledgePathRepository({ + maxListLimit: 20, + maxPaths: 20, + }), + multimodalManifests: checkpointManifests, + objectStorage: adapter.objectStorage, + outlineBuilder, + outlineSummaryEnhancer: { + enhance: async () => { + summaryCalls += 1; + throw new Error("outline summary must not be regenerated"); + }, + }, + outlines, + pageIndexBuild: { + materializeBuilding: async () => { + throw new Error("PageIndex must not be regenerated"); + }, + }, + parser: { + kind: "native-markdown", + parse: async () => { + parserCalls += 1; + throw new Error("document must not be reparsed"); + }, + }, + reindexer: { + getCanonicalArtifact: async (input) => + canonicalArtifactAvailable ? artifacts.getByDocumentVersion(input) : null, + reindex: async (input) => { + resetFailedProjectionFlags.push(input.resetFailedProjections); + return { + artifact: input.parseArtifact, + nodeIds: ["018f0d60-7a49-7cc2-9c1b-5b36f18f7005"], + nodesCreated: 1, + projectionIds: ["018f0d60-7a49-7cc2-9c1b-5b36f18f7006"], + projectionsCreated: 1, + status: "rebuilt", + }; + }, + }, + }); + + await expect(worker.process(payload)).rejects.toThrow("parse artifact is missing"); + canonicalArtifactAvailable = true; + await expect(worker.process(payload)).rejects.toThrow("derived components are missing"); + manifestCheckpoint = "invalid"; + await expect(worker.process(payload)).rejects.toThrow("derived component lineage is invalid"); + manifestCheckpoint = "valid"; + await expect(worker.process(payload)).resolves.toMatchObject({ stage: "projection_built" }); + expect(parserCalls).toBe(0); + expect(summaryCalls).toBe(0); + expect(resetFailedProjectionFlags).toEqual([true]); + expect(receipts).toEqual([ + expect.objectContaining({ + componentReceipt: expect.objectContaining({ + documentOutlines: [{ componentKey: persistedOutline.id, generationId }], + multimodalManifests: [{ componentKey: persistedManifest.id, generationId }], + }), + }), + ]); + }); + it("builds retry derivatives from the canonical artifact returned by reindexing", async () => { const adapter = createNodePlatformAdapter({ env: {} }); const assets = createInMemoryDocumentAssetRepository({ diff --git a/knowledge-fs/packages/api/src/document-compilation-worker.ts b/knowledge-fs/packages/api/src/document-compilation-worker.ts index 49baa10040d..a34aeb2713f 100644 --- a/knowledge-fs/packages/api/src/document-compilation-worker.ts +++ b/knowledge-fs/packages/api/src/document-compilation-worker.ts @@ -1,9 +1,14 @@ import { z } from "@hono/zod-openapi"; import type { ChunkConfig } from "@knowledge/compute"; import { + type DocumentAsset, + type DocumentMultimodalManifest, + type DocumentOutline, type JobPayload, + type KnowledgePath, type KnowledgeSpaceEmbeddingProfile, type KnowledgeSpaceRetrievalProfile, + type ParseArtifact, type PlatformAdapter, PublicationGenerationIdSchema, TenantIdSchema, @@ -291,56 +296,85 @@ export function createDocumentCompilationWorker({ }) : objectStorage; const compile = async () => { - const body = await objectStorage.getObject(activeAsset.objectKey); - - if (!body) { - throw new Error("Document compilation object not found"); + const initialJob = await jobs.get(input.documentCompilationJobId); + if (!initialJob) { + throw new Error("Document compilation job not found"); } + const resumeParsedGeneration = + publicationGenerationId !== undefined && + hasReachedCompilationStage(initialJob.stage, "parsed"); + const resumeOutlineGeneration = + publicationGenerationId !== undefined && + hasReachedCompilationStage(initialJob.stage, "outline_built"); + let canonicalArtifact: ParseArtifact; + if (resumeParsedGeneration) { + if (!reindexer.getCanonicalArtifact) { + throw new Error( + `Document compilation checkpoint=${initialJob.stage} cannot load its parse artifact`, + ); + } + const persistedArtifact = await reindexer.getCanonicalArtifact({ + documentAssetId: activeAsset.id, + version: activeAsset.version, + }); + if (!persistedArtifact) { + throw new Error( + `Document compilation checkpoint=${initialJob.stage} parse artifact is missing`, + ); + } + canonicalArtifact = persistedArtifact; + } else { + const body = await objectStorage.getObject(activeAsset.objectKey); - const parsedArtifact = await parser.parse({ - body, - documentAssetId: activeAsset.id, - filename: activeAsset.filename, - mimeType: activeAsset.mimeType, - ...(signal ? { signal } : {}), - version: activeAsset.version, - }); - await assertWritable(); - const rasterized = await rasterizeDocumentPdfMultimodalAssets({ - artifact: parsedArtifact, - documentBody: body, - documentMimeType: activeAsset.mimeType, - knowledgeSpaceId: input.knowledgeSpaceId, - ...(multimodalMaxPdfRasterizedAssets - ? { maxRasterizedAssets: multimodalMaxPdfRasterizedAssets } - : {}), - objectStorage: multimodalObjectStorage, - ...(pdfRasterizer ? { rasterizer: pdfRasterizer } : {}), - tenantId: input.tenantId, - }); - await assertWritable(); - const { artifact } = await extractDocumentMultimodalAssets({ - ...(multimodalLocalAssetAllowlist - ? { allowLocalAssetPaths: multimodalLocalAssetAllowlist } - : {}), - artifact: rasterized.artifact, - knowledgeSpaceId: input.knowledgeSpaceId, - ...(multimodalMaxExtractedAssets - ? { maxExtractedAssets: multimodalMaxExtractedAssets } - : {}), - ...(multimodalMaxLocalAssetBytes - ? { maxLocalAssetBytes: multimodalMaxLocalAssetBytes } - : {}), - ...(multimodalImageVariantGenerator - ? { imageVariantGenerator: multimodalImageVariantGenerator } - : {}), - objectStorage: multimodalObjectStorage, - tenantId: input.tenantId, - }); - await assertWritable(); - const canonicalArtifact = reindexer.canonicalizeArtifact - ? await reindexer.canonicalizeArtifact(artifact) - : artifact; + if (!body) { + throw new Error("Document compilation object not found"); + } + + const parsedArtifact = await parser.parse({ + body, + documentAssetId: activeAsset.id, + filename: activeAsset.filename, + mimeType: activeAsset.mimeType, + ...(signal ? { signal } : {}), + version: activeAsset.version, + }); + await assertWritable(); + const rasterized = await rasterizeDocumentPdfMultimodalAssets({ + artifact: parsedArtifact, + documentBody: body, + documentMimeType: activeAsset.mimeType, + knowledgeSpaceId: input.knowledgeSpaceId, + ...(multimodalMaxPdfRasterizedAssets + ? { maxRasterizedAssets: multimodalMaxPdfRasterizedAssets } + : {}), + objectStorage: multimodalObjectStorage, + ...(pdfRasterizer ? { rasterizer: pdfRasterizer } : {}), + tenantId: input.tenantId, + }); + await assertWritable(); + const { artifact } = await extractDocumentMultimodalAssets({ + ...(multimodalLocalAssetAllowlist + ? { allowLocalAssetPaths: multimodalLocalAssetAllowlist } + : {}), + artifact: rasterized.artifact, + knowledgeSpaceId: input.knowledgeSpaceId, + ...(multimodalMaxExtractedAssets + ? { maxExtractedAssets: multimodalMaxExtractedAssets } + : {}), + ...(multimodalMaxLocalAssetBytes + ? { maxLocalAssetBytes: multimodalMaxLocalAssetBytes } + : {}), + ...(multimodalImageVariantGenerator + ? { imageVariantGenerator: multimodalImageVariantGenerator } + : {}), + objectStorage: multimodalObjectStorage, + tenantId: input.tenantId, + }); + await assertWritable(); + canonicalArtifact = reindexer.canonicalizeArtifact + ? await reindexer.canonicalizeArtifact(artifact) + : artifact; + } const documentIndexOverrides = indexOverrides ? await indexOverrides.resolve({ compilationAttemptId: input.documentCompilationJobId, @@ -350,95 +384,98 @@ export function createDocumentCompilationWorker({ tenantId: input.tenantId, }) : {}; - await assertWritable(); - await jobs.advance(input.documentCompilationJobId, "parsed"); - const multimodalManifest = createDocumentMultimodalManifestBuilder().build({ - artifact: canonicalArtifact, - knowledgeSpaceId: input.knowledgeSpaceId, - ...(publicationGenerationId ? { publicationGenerationId } : {}), - }); let documentOutlineIds: readonly string[] = []; let knowledgePathIds: readonly string[] = []; - if (outlineBuilder && outlines) { - const deterministicOutline = outlineBuilder.build({ + let persistedManifest: DocumentMultimodalManifest; + if (resumeOutlineGeneration && publicationGenerationId) { + const [persistedOutline, resumedManifest] = await Promise.all([ + outlines?.getByDocumentVersion({ + documentAssetId: activeAsset.id, + publicationGenerationId, + version: activeAsset.version, + }), + multimodalManifests.getByDocumentVersion({ + documentAssetId: activeAsset.id, + publicationGenerationId, + version: activeAsset.version, + }), + ]); + if (!persistedOutline || !resumedManifest) { + throw new Error( + `Document compilation checkpoint=${initialJob.stage} derived components are missing`, + ); + } + assertResumableCompilationComponents({ + artifact: canonicalArtifact, + asset: activeAsset, + manifest: resumedManifest, + outline: persistedOutline, + publicationGenerationId, + }); + persistedManifest = resumedManifest; + documentOutlineIds = [persistedOutline.id]; + knowledgePathIds = buildCompilationKnowledgePaths({ + asset: activeAsset, + generateId: () => publicationGenerationId, + manifest: persistedManifest, + outline: persistedOutline, + publicationGenerationId, + tenantId: input.tenantId, + }).map((path) => path.id); + } else { + await assertWritable(); + await jobs.advance(input.documentCompilationJobId, "parsed"); + const multimodalManifest = createDocumentMultimodalManifestBuilder().build({ + artifact: canonicalArtifact, knowledgeSpaceId: input.knowledgeSpaceId, - parseArtifact: canonicalArtifact, ...(publicationGenerationId ? { publicationGenerationId } : {}), }); - const outline = outlineSummaryEnhancer - ? await outlineSummaryEnhancer.enhance({ - outline: deterministicOutline, - parseArtifact: canonicalArtifact, - ...(frozenRetrievalProfile ? { retrievalProfile: frozenRetrievalProfile } : {}), - ...(signal ? { signal } : {}), - tenantId: input.tenantId, - }) - : deterministicOutline; - await assertWritable(); - const persistedOutline = await outlines.upsert(outline); - if (publicationGenerationId && documentIndexOverrides.enablePageIndex !== false) { - await assertWritable(); - await pageIndexBuild?.materializeBuilding({ - builtAt: persistedOutline.updatedAt ?? persistedOutline.createdAt, - outline: persistedOutline, - tenantId: input.tenantId, + if (outlineBuilder && outlines) { + const deterministicOutline = outlineBuilder.build({ + knowledgeSpaceId: input.knowledgeSpaceId, + parseArtifact: canonicalArtifact, + ...(publicationGenerationId ? { publicationGenerationId } : {}), }); - } - documentOutlineIds = [persistedOutline.id]; - if (knowledgePaths && generateKnowledgePathId) { + const outline = outlineSummaryEnhancer + ? await outlineSummaryEnhancer.enhance({ + outline: deterministicOutline, + parseArtifact: canonicalArtifact, + ...(frozenRetrievalProfile ? { retrievalProfile: frozenRetrievalProfile } : {}), + ...(signal ? { signal } : {}), + tenantId: input.tenantId, + }) + : deterministicOutline; await assertWritable(); - const persistedPaths = await knowledgePaths.upsertMany([ - ...(publicationGenerationId - ? [ - buildDocumentKnowledgePath({ - asset: activeAsset, - id: generateKnowledgePathId(), - publicationGenerationId, - tenantId: input.tenantId, - }), - ] - : []), - buildDocumentMultimodalManifestKnowledgePath({ - asset: activeAsset, - id: generateKnowledgePathId(), - ...(publicationGenerationId ? { publicationGenerationId } : {}), - tenantId: input.tenantId, - }), - ...buildDocumentMultimodalAssetKnowledgePaths({ - asset: activeAsset, - generateId: generateKnowledgePathId, - manifest: multimodalManifest, - ...(publicationGenerationId ? { publicationGenerationId } : {}), - tenantId: input.tenantId, - }), - ...buildDocumentMultimodalResourceKnowledgePaths({ - asset: activeAsset, - generateId: generateKnowledgePathId, - manifest: multimodalManifest, - ...(publicationGenerationId ? { publicationGenerationId } : {}), - tenantId: input.tenantId, - }), - buildDocumentOutlineKnowledgePath({ - asset: activeAsset, - id: generateKnowledgePathId(), - ...(publicationGenerationId ? { publicationGenerationId } : {}), - tenantId: input.tenantId, - }), - ...buildDocumentSectionKnowledgePaths({ - asset: activeAsset, - generateId: generateKnowledgePathId, + const persistedOutline = await outlines.upsert(outline); + if (publicationGenerationId && documentIndexOverrides.enablePageIndex !== false) { + await assertWritable(); + await pageIndexBuild?.materializeBuilding({ + builtAt: persistedOutline.updatedAt ?? persistedOutline.createdAt, outline: persistedOutline, - ...(publicationGenerationId ? { publicationGenerationId } : {}), tenantId: input.tenantId, - }), - ]); - knowledgePathIds = persistedPaths.map((path) => path.id); + }); + } + documentOutlineIds = [persistedOutline.id]; + if (knowledgePaths && generateKnowledgePathId) { + await assertWritable(); + const persistedPaths = await knowledgePaths.upsertMany( + buildCompilationKnowledgePaths({ + asset: activeAsset, + generateId: generateKnowledgePathId, + manifest: multimodalManifest, + outline: persistedOutline, + ...(publicationGenerationId ? { publicationGenerationId } : {}), + tenantId: input.tenantId, + }), + ); + knowledgePathIds = persistedPaths.map((path) => path.id); + } } + await assertWritable(); + persistedManifest = await multimodalManifests.upsert(multimodalManifest); + await assertWritable(); + await jobs.advance(input.documentCompilationJobId, "outline_built"); } - await assertWritable(); - const persistedManifest = await multimodalManifests.upsert(multimodalManifest); - await assertWritable(); - await jobs.advance(input.documentCompilationJobId, "outline_built"); const resolvedEmbedding = frozenEmbeddingProfile ? frozenEmbeddingProfile @@ -473,6 +510,7 @@ export function createDocumentCompilationWorker({ publicationGenerationId || legacyStagedProjectionPublication ? "building" : "ready", projectionVersion: input.version, ...(publicationGenerationId ? { publicationGenerationId } : {}), + ...(initialJob.stage === "outline_built" ? { resetFailedProjections: true } : {}), ...(signal ? { signal } : {}), tenantId: input.tenantId, ...(visualEmbeddingModel ? { visualModel: visualEmbeddingModel } : {}), @@ -766,6 +804,121 @@ function createDeletionFencedCompilationObjectStorage({ }; } +const resumableCompilationStages: readonly DocumentCompilationJob["stage"][] = [ + "queued", + "parsed", + "outline_built", + "nodes_generated", + "projection_built", + "smoke_eval_passed", + "published", +]; + +function hasReachedCompilationStage( + current: DocumentCompilationJob["stage"], + expected: DocumentCompilationJob["stage"], +): boolean { + const currentIndex = resumableCompilationStages.indexOf(current); + const expectedIndex = resumableCompilationStages.indexOf(expected); + + return currentIndex >= expectedIndex && expectedIndex >= 0; +} + +function assertResumableCompilationComponents({ + artifact, + asset, + manifest, + outline, + publicationGenerationId, +}: { + readonly artifact: ParseArtifact; + readonly asset: DocumentAsset; + readonly manifest: DocumentMultimodalManifest; + readonly outline: DocumentOutline; + readonly publicationGenerationId: string; +}): void { + const hasArtifactLineage = + artifact.documentAssetId === asset.id && + artifact.version === asset.version && + outline.documentAssetId === asset.id && + outline.knowledgeSpaceId === asset.knowledgeSpaceId && + outline.version === asset.version && + outline.parseArtifactId === artifact.id && + outline.artifactHash === artifact.artifactHash && + outline.publicationGenerationId === publicationGenerationId && + manifest.documentAssetId === asset.id && + manifest.knowledgeSpaceId === asset.knowledgeSpaceId && + manifest.version === asset.version && + manifest.parseArtifactId === artifact.id && + manifest.artifactHash === artifact.artifactHash && + manifest.publicationGenerationId === publicationGenerationId; + if (!hasArtifactLineage) { + throw new Error("Document compilation checkpoint derived component lineage is invalid"); + } +} + +function buildCompilationKnowledgePaths({ + asset, + generateId, + manifest, + outline, + publicationGenerationId, + tenantId, +}: { + readonly asset: DocumentAsset; + readonly generateId: () => string; + readonly manifest: DocumentMultimodalManifest; + readonly outline: DocumentOutline; + readonly publicationGenerationId?: string | undefined; + readonly tenantId: string; +}): KnowledgePath[] { + return [ + ...(publicationGenerationId + ? [ + buildDocumentKnowledgePath({ + asset, + id: generateId(), + publicationGenerationId, + tenantId, + }), + ] + : []), + buildDocumentMultimodalManifestKnowledgePath({ + asset, + id: generateId(), + ...(publicationGenerationId ? { publicationGenerationId } : {}), + tenantId, + }), + ...buildDocumentMultimodalAssetKnowledgePaths({ + asset, + generateId, + manifest, + ...(publicationGenerationId ? { publicationGenerationId } : {}), + tenantId, + }), + ...buildDocumentMultimodalResourceKnowledgePaths({ + asset, + generateId, + manifest, + ...(publicationGenerationId ? { publicationGenerationId } : {}), + tenantId, + }), + buildDocumentOutlineKnowledgePath({ + asset, + id: generateId(), + ...(publicationGenerationId ? { publicationGenerationId } : {}), + tenantId, + }), + ...buildDocumentSectionKnowledgePaths({ + asset, + generateId, + outline, + ...(publicationGenerationId ? { publicationGenerationId } : {}), + tenantId, + }), + ]; +} + function componentReferences( componentKeys: readonly string[], generationId: string, diff --git a/knowledge-fs/packages/api/src/index-reindexer.test.ts b/knowledge-fs/packages/api/src/index-reindexer.test.ts index 5a8debd4ae6..080367bc1d5 100644 --- a/knowledge-fs/packages/api/src/index-reindexer.test.ts +++ b/knowledge-fs/packages/api/src/index-reindexer.test.ts @@ -8,6 +8,7 @@ import { import { describe, expect, it } from "vitest"; import { + createDenseVectorProjectionBuilder, createFtsProjectionBuilder, createVisualEmbeddingProjectionBuilder, } from "./index-projection-builders"; @@ -496,6 +497,140 @@ describe("incremental reindexer", () => { ).resolves.toMatchObject({ building: 0, failed: 1, ready: 0, total: 1 }); }); + it("replaces failed generation projections when retrying from the outline checkpoint", async () => { + const artifacts = createInMemoryParseArtifactRepository({ maxArtifacts: 4 }); + const nodes = createInMemoryKnowledgeNodeRepository({ + maxBatchSize: 4, + maxListLimit: 4, + maxNodes: 4, + }); + const projections = createInMemoryIndexProjectionRepository({ + maxBatchSize: 4, + maxListLimit: 4, + maxProjections: 8, + }); + let embeddingCalls = 0; + const reindexer = createIncrementalReindexer({ + artifacts, + compute: computeRuntime(), + denseBuilder: createDenseVectorProjectionBuilder({ + embeddings: { + embed: async () => { + embeddingCalls += 1; + if (embeddingCalls === 1) { + throw new Error("dify model runtime timeout"); + } + return { + dense: [[0.25, 0.75]], + metadata: { dimension: 2, model: "dense-v1", provider: "dify-model-runtime" }, + model: "dense-v1", + }; + }, + kind: "dify-model-runtime", + models: async () => [], + }, + maxBatchSize: 4, + projections, + }), + ftsBuilder: createFtsProjectionBuilder({ maxBatchSize: 4, projections }), + maxNodes: 4, + nodes, + projections, + }); + const signal = new AbortController().signal; + const input = { + chunkConfig: { maxChunkChars: 512, overlapChars: 64 }, + denseModel: "dense-v1", + knowledgeSpaceId: KNOWLEDGE_SPACE_ID, + language: "zh-CN", + parseArtifact: parseArtifact(), + projectionVersion: 1, + publicationGenerationId: PUBLICATION_GENERATION_ID, + signal, + tenantId: "tenant-1", + } as const; + + await expect(reindexer.reindex(input)).rejects.toThrow("dify model runtime timeout"); + await expect( + reindexer.getCanonicalArtifact?.({ + documentAssetId: DOCUMENT_ASSET_ID, + version: 1, + }), + ).resolves.toMatchObject({ id: parseArtifact().id }); + await expect( + reindexer.getCanonicalArtifact?.({ + documentAssetId: "018f0d60-7a49-7cc2-9c1b-5b36f18f2cff", + version: 1, + }), + ).resolves.toBeNull(); + await expect( + projections.summarizeVersion({ + knowledgeSpaceId: KNOWLEDGE_SPACE_ID, + projectionVersion: 1, + publicationGenerationId: PUBLICATION_GENERATION_ID, + type: "fts", + }), + ).resolves.toMatchObject({ building: 0, failed: 1, total: 1 }); + + await expect( + reindexer.reindex({ + ...input, + resetFailedProjections: true, + }), + ).resolves.toMatchObject({ + nodesCreated: 1, + projectionsCreated: 2, + status: "rebuilt", + }); + await expect( + projections.summarizeVersion({ + knowledgeSpaceId: KNOWLEDGE_SPACE_ID, + projectionVersion: 1, + publicationGenerationId: PUBLICATION_GENERATION_ID, + type: "fts", + }), + ).resolves.toMatchObject({ building: 1, failed: 0, total: 1 }); + await expect( + projections.summarizeVersion({ + knowledgeSpaceId: KNOWLEDGE_SPACE_ID, + projectionVersion: 1, + publicationGenerationId: PUBLICATION_GENERATION_ID, + type: "dense-vector", + }), + ).resolves.toMatchObject({ building: 1, failed: 0, total: 1 }); + expect(embeddingCalls).toBe(2); + }); + + it("requires a generation and projection repository before resetting failed projections", async () => { + const baseOptions = { + artifacts: createInMemoryParseArtifactRepository({ maxArtifacts: 4 }), + compute: computeRuntime(), + maxNodes: 4, + nodes: createInMemoryKnowledgeNodeRepository({ + maxBatchSize: 4, + maxListLimit: 4, + maxNodes: 4, + }), + }; + const reindexer = createIncrementalReindexer(baseOptions); + const input = { + knowledgeSpaceId: KNOWLEDGE_SPACE_ID, + parseArtifact: parseArtifact(), + projectionVersion: 1, + resetFailedProjections: true, + } as const; + + await expect(reindexer.reindex(input)).rejects.toThrow( + "can reset failed projections only for a publication generation", + ); + await expect( + reindexer.reindex({ + ...input, + publicationGenerationId: PUBLICATION_GENERATION_ID, + }), + ).rejects.toThrow("requires a projection repository to reset failed projections"); + }); + it("can fail the whole candidate after a batched publication throws partway through", async () => { const statuses = new Map([ ["projection-a", "building"], diff --git a/knowledge-fs/packages/api/src/index-reindexer.ts b/knowledge-fs/packages/api/src/index-reindexer.ts index 3f26e8f4196..b349e6068fc 100644 --- a/knowledge-fs/packages/api/src/index-reindexer.ts +++ b/knowledge-fs/packages/api/src/index-reindexer.ts @@ -20,7 +20,11 @@ import type { IndexProjectionRepository } from "./index-projection-repository"; import { isPlainObject } from "./json-utils"; import type { KnowledgeFsOperationLeaseCoordinator } from "./knowledge-fs-operation-leases"; import { type KnowledgeNodeRepository, cloneKnowledgeNode } from "./knowledge-node-repository"; -import { type ParseArtifactRepository, cloneParseArtifact } from "./parse-artifact-repository"; +import { + type ParseArtifactLookupInput, + type ParseArtifactRepository, + cloneParseArtifact, +} from "./parse-artifact-repository"; export interface IncrementalReindexInput { readonly chunkConfig?: ChunkConfig | undefined; @@ -37,6 +41,8 @@ export interface IncrementalReindexInput { readonly projectionStatus?: ProjectionBuildStatus | undefined; readonly projectionVersion: number; readonly publicationGenerationId?: string | undefined; + /** Removes failed projections from an unpublished generation before rebuilding a retry. */ + readonly resetFailedProjections?: boolean | undefined; readonly signal?: AbortSignal | undefined; readonly tenantId?: string | undefined; readonly visualModel?: string | undefined; @@ -67,6 +73,7 @@ export interface UpdateIncrementalReindexProjectionStatusInput { export interface IncrementalReindexer { canonicalizeArtifact?(artifact: ParseArtifact): Promise; failProjections?(input: UpdateIncrementalReindexProjectionStatusInput): Promise; + getCanonicalArtifact?(input: ParseArtifactLookupInput): Promise; publishProjections?(input: UpdateIncrementalReindexProjectionStatusInput): Promise; reindex(input: IncrementalReindexInput): Promise; } @@ -140,6 +147,13 @@ export function createIncrementalReindexer({ cloneParseArtifact( await artifacts.create(cloneParseArtifact(ParseArtifactSchema.parse(artifact))), ), + getCanonicalArtifact: async (input: ParseArtifactLookupInput) => { + const persisted = await artifacts.getByDocumentVersion(input); + + return persisted + ? cloneParseArtifact(await artifacts.create(cloneParseArtifact(persisted))) + : null; + }, ...(canUpdateProjectionStatuses ? { failProjections: async (input: UpdateIncrementalReindexProjectionStatusInput) => { @@ -167,6 +181,16 @@ export function createIncrementalReindexer({ input.publicationGenerationId === undefined ? undefined : PublicationGenerationIdSchema.parse(input.publicationGenerationId); + if (input.resetFailedProjections && !publicationGenerationId) { + throw new Error( + "Incremental reindexer can reset failed projections only for a publication generation", + ); + } + if (input.resetFailedProjections && !projections) { + throw new Error( + "Incremental reindexer requires a projection repository to reset failed projections", + ); + } const reindex = async (): Promise => { input.signal?.throwIfAborted(); const storedArtifact = await artifacts.create(parseArtifact); @@ -206,6 +230,16 @@ export function createIncrementalReindexer({ ? await nodes.upsertMany(chunkedNodes.map(cloneKnowledgeNode)) : []; input.signal?.throwIfAborted(); + if (input.resetFailedProjections && projections) { + for (const nodeBatch of chunkNodes(storedNodes, projectionBatchSize)) { + await projections.deleteByNodeIds({ + knowledgeSpaceId: input.knowledgeSpaceId, + maxProjections: nodeBatch.length * 3, + nodeIds: nodeBatch.map((node) => node.id), + }); + } + } + input.signal?.throwIfAborted(); const projectionIds: string[] = []; const observedVectorSpaces = new Map< string, diff --git a/knowledge-fs/packages/api/src/parse-artifact-repository.test.ts b/knowledge-fs/packages/api/src/parse-artifact-repository.test.ts index a7b9adf6ee8..e0d1b93f940 100644 --- a/knowledge-fs/packages/api/src/parse-artifact-repository.test.ts +++ b/knowledge-fs/packages/api/src/parse-artifact-repository.test.ts @@ -34,6 +34,78 @@ const artifact = ParseArtifactSchema.parse({ }) satisfies ParseArtifact; describe("parse artifact repositories", () => { + it("keeps generated element ids bound to the first persisted artifact on retry", async () => { + const retryArtifactId = "018f0d60-7a49-7cc2-9c1b-5b36f18f2c46"; + const first = ParseArtifactSchema.parse({ + ...artifact, + elements: [ + { + ...artifact.elements[0], + id: `${artifact.id}:element-1`, + }, + ], + }); + const retry = ParseArtifactSchema.parse({ + ...first, + createdAt: "2026-05-09T11:01:01.000Z", + elements: [ + { + ...first.elements[0], + id: `${retryArtifactId}:element-1`, + }, + ], + id: retryArtifactId, + }); + const memory = createInMemoryParseArtifactRepository({ maxArtifacts: 2 }); + const fake = createFakeParseArtifactExecutor(); + const database = createDatabaseParseArtifactRepository({ + database: createSchemaDatabaseAdapter({ executor: fake.executor, kind: "postgres" }), + }); + + for (const repository of [memory, database]) { + await repository.create(first); + + await expect(repository.create(retry)).resolves.toMatchObject({ + createdAt: first.createdAt, + elements: [{ id: `${first.id}:element-1` }], + id: first.id, + }); + await expect( + repository.getByDocumentVersion({ + documentAssetId: first.documentAssetId, + version: first.version, + }), + ).resolves.toMatchObject({ + elements: [{ id: `${first.id}:element-1` }], + id: first.id, + }); + await expect(repository.getById({ id: retryArtifactId })).resolves.toBeNull(); + } + }); + + it("fails closed when a concurrent row change prevents generated id repair", async () => { + const retryArtifactId = "018f0d60-7a49-7cc2-9c1b-5b36f18f2c46"; + const first = ParseArtifactSchema.parse({ + ...artifact, + elements: [{ ...artifact.elements[0], id: `${artifact.id}:element-1` }], + }); + const fake = createFakeParseArtifactExecutor({ failUpdates: true }); + const repository = createDatabaseParseArtifactRepository({ + database: createSchemaDatabaseAdapter({ executor: fake.executor, kind: "postgres" }), + }); + await repository.create(first); + + const retry = ParseArtifactSchema.parse({ + ...first, + elements: [{ ...first.elements[0], id: `${retryArtifactId}:element-1` }], + id: retryArtifactId, + }); + + await expect(repository.create(retry)).rejects.toThrow( + "generated element ids could not be canonicalized", + ); + }); + it("stores clone-isolated artifacts and bounds in-memory capacity", async () => { const repository = createInMemoryParseArtifactRepository({ maxArtifacts: 1 }); @@ -205,7 +277,9 @@ describe("parse artifact repositories", () => { }); }); -function createFakeParseArtifactExecutor() { +function createFakeParseArtifactExecutor({ + failUpdates = false, +}: { readonly failUpdates?: boolean } = {}) { const calls: DatabaseExecuteInput[] = []; const rows = new Map(); const executor = async (input: DatabaseExecuteInput): Promise => { @@ -238,9 +312,20 @@ function createFakeParseArtifactExecutor() { version: Number(version), } satisfies DatabaseRow; - rows.set(`${row.document_asset_id}:${row.version}`, row); + const key = `${row.document_asset_id}:${row.version}`; + const existing = rows.get(key); + rows.set( + key, + existing + ? { + ...row, + created_at: existing.created_at, + id: existing.id, + } + : row, + ); - return { rows: [{ ...row }], rowsAffected: 1 }; + return { rows: [{ ...(rows.get(key) ?? row) }], rowsAffected: 1 }; } if (input.operation === "select") { @@ -253,6 +338,24 @@ function createFakeParseArtifactExecutor() { return { rows: row ? [{ ...row }] : [], rowsAffected: row ? 1 : 0 }; } + if (input.operation === "update") { + if (failUpdates) { + return { rows: [], rowsAffected: 0 }; + } + const [elements, id, documentAssetId, version] = input.params; + const key = `${String(documentAssetId)}:${Number(version)}`; + const row = rows.get(key); + if (!row || row.id !== String(id)) { + return { rows: [], rowsAffected: 0 }; + } + rows.set(key, { + ...row, + elements: typeof elements === "string" ? JSON.parse(elements) : elements, + }); + + return { rows: [], rowsAffected: 1 }; + } + return { rows: [], rowsAffected: 0 }; }; diff --git a/knowledge-fs/packages/api/src/parse-artifact-repository.ts b/knowledge-fs/packages/api/src/parse-artifact-repository.ts index 19c1a1e41db..84df2229b74 100644 --- a/knowledge-fs/packages/api/src/parse-artifact-repository.ts +++ b/knowledge-fs/packages/api/src/parse-artifact-repository.ts @@ -96,6 +96,30 @@ export function cloneParseArtifact(artifact: ParseArtifact): ParseArtifact { return ParseArtifactSchema.parse(JSON.parse(JSON.stringify(artifact)) as unknown); } +function bindGeneratedElementIdsToArtifact( + artifact: ParseArtifact, + canonicalArtifactId: string, +): ParseArtifact { + const generatedElementIds = artifact.elements.every((element, index) => { + const match = element.id.match( + /^[0-9a-f]{8}-[0-9a-f]{4}-[1-8][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}:element-(\d+)$/iu, + ); + + return match?.[1] === String(index + 1); + }); + + return cloneParseArtifact({ + ...artifact, + elements: generatedElementIds + ? artifact.elements.map((element, index) => ({ + ...element, + id: `${canonicalArtifactId}:element-${index + 1}`, + })) + : artifact.elements, + id: canonicalArtifactId, + }); +} + export function createInMemoryParseArtifactRepository({ maxArtifacts, }: InMemoryParseArtifactRepositoryOptions): ParseArtifactRepository { @@ -115,9 +139,10 @@ export function createInMemoryParseArtifactRepository({ throw new ParseArtifactCapacityExceededError(maxArtifacts); } - const stored = existing - ? cloneParseArtifact({ ...artifact, createdAt: existing.createdAt, id: existing.id }) - : artifact; + const stored = bindGeneratedElementIdsToArtifact( + existing ? { ...artifact, createdAt: existing.createdAt } : artifact, + existing?.id ?? artifact.id, + ); artifacts.set(key, stored); return cloneParseArtifact(stored); @@ -176,6 +201,46 @@ export function createDatabaseParseArtifactRepository({ database, }: DatabaseParseArtifactRepositoryOptions): ParseArtifactRepository { const tableName = "parse_artifacts"; + const repairGeneratedElementIds = async (persisted: ParseArtifact): Promise => { + const canonical = bindGeneratedElementIdsToArtifact(persisted, persisted.id); + if (JSON.stringify(canonical.elements) === JSON.stringify(persisted.elements)) { + return canonical; + } + + const repaired = await database.execute({ + maxRows: 1, + operation: "update", + params: [ + JSON.stringify(canonical.elements), + canonical.id, + canonical.documentAssetId, + canonical.version, + ], + sql: `UPDATE ${quoteDatabaseIdentifier(database, tableName)} SET ${quoteDatabaseIdentifier( + database, + "elements", + )} = ${jsonInsertPlaceholder( + database, + 1, + "elements", + )} WHERE ${quoteDatabaseIdentifier(database, "id")} = ${databasePlaceholder( + database, + 2, + )} AND ${quoteDatabaseIdentifier( + database, + "document_asset_id", + )} = ${databasePlaceholder(database, 3)} AND ${quoteDatabaseIdentifier( + database, + "version", + )} = ${databasePlaceholder(database, 4)};`, + tableName, + }); + if (repaired.rowsAffected !== 1) { + throw new Error("Parse artifact generated element ids could not be canonicalized"); + } + + return canonical; + }; return { create: async (input) => { @@ -247,7 +312,7 @@ export function createDatabaseParseArtifactRepository({ }); if (result.rows[0]) { - return mapParseArtifactRow(result.rows[0]); + return repairGeneratedElementIds(mapParseArtifactRow(result.rows[0])); } const stored = await database.execute({ @@ -283,7 +348,7 @@ export function createDatabaseParseArtifactRepository({ throw new Error("Parse artifact upsert resolved a mismatched persisted logical row"); } - return persisted; + return repairGeneratedElementIds(persisted); }, getByDocumentVersion: async ({ documentAssetId, version }) => { const result = await database.execute({