From 0c46afba59fa44f50ef9516415333fce862f43fc Mon Sep 17 00:00:00 2001 From: lofyer Date: Thu, 13 Aug 2026 04:47:44 +0800 Subject: [PATCH] fix: preserve knowledge indexing state --- src/main/knowledge/knowledge-database.test.ts | 182 +++++++++++- src/main/knowledge/knowledge-database.ts | 220 +++++++++++++-- src/main/knowledge/knowledge-service.test.ts | 260 ++++++++++++++++++ src/main/knowledge/knowledge-service.ts | 201 ++++++++++---- 4 files changed, 772 insertions(+), 91 deletions(-) diff --git a/src/main/knowledge/knowledge-database.test.ts b/src/main/knowledge/knowledge-database.test.ts index ab87665..9cb927b 100644 --- a/src/main/knowledge/knowledge-database.test.ts +++ b/src/main/knowledge/knowledge-database.test.ts @@ -86,7 +86,7 @@ describe('KnowledgeDatabase', () => { const inspection = new DatabaseSync(path) expect( inspection.prepare('PRAGMA user_version').get() - ).toEqual({ user_version: 9 }) + ).toEqual({ user_version: 10 }) expect( inspection .prepare('SELECT version FROM schema_migrations ORDER BY version') @@ -100,7 +100,8 @@ describe('KnowledgeDatabase', () => { { version: 6 }, { version: 7 }, { version: 8 }, - { version: 9 } + { version: 9 }, + { version: 10 } ]) inspection.close() @@ -192,7 +193,7 @@ describe('KnowledgeDatabase', () => { }) }) - it('upgrades an existing v1 database through knowledge schema v9', async () => { + it('upgrades an existing v1 database through knowledge schema v10', async () => { const { database, path } = await createDatabase() const knowledgeBase = database.createKnowledgeBase({ name: 'Version one data', @@ -208,7 +209,8 @@ describe('KnowledgeDatabase', () => { DROP TABLE embedding_index_state; DROP TABLE chunk_embeddings; DROP TABLE knowledge_tasks; - DELETE FROM schema_migrations WHERE version IN (2, 3, 4, 5, 6, 7, 8, 9); + DELETE FROM schema_migrations + WHERE version IN (2, 3, 4, 5, 6, 7, 8, 9, 10); PRAGMA user_version = 1; `) downgrade.close() @@ -218,7 +220,7 @@ describe('KnowledgeDatabase', () => { upgraded.initialize() const inspection = new DatabaseSync(path) expect(inspection.prepare('PRAGMA user_version').get()).toEqual({ - user_version: 9 + user_version: 10 }) expect( inspection @@ -325,7 +327,7 @@ describe('KnowledgeDatabase', () => { WHERE new.enabled = 1 AND new.role <> 'parent'; END; INSERT INTO chunks_fts(chunks_fts) VALUES ('rebuild'); - DELETE FROM schema_migrations WHERE version = 9; + DELETE FROM schema_migrations WHERE version IN (9, 10); PRAGMA user_version = 8; `) downgrade @@ -478,6 +480,38 @@ describe('KnowledgeDatabase', () => { ).toMatchObject({ status: 'interrupted', progress: 30 }) }) + it('bounds task status text from runtime failures', async () => { + const { database } = await createDatabase() + const library = database.createKnowledgeBase({ + name: 'Bounded task status', + storageMode: 'reference' + }) + const task = database.createKnowledgeTask({ + libraryId: library.id, + documentName: library.name, + scope: 'library', + kind: 'library-rebuild' + }) + + expect( + database.updateKnowledgeTask(task.id, { + status: 'failed', + message: 'm'.repeat(2_000), + error: { + message: 'e'.repeat(2_000), + remedy: 'r'.repeat(2_000) + } + }) + ).toMatchObject({ + status: 'failed', + message: 'm'.repeat(1_000), + error: { + message: 'e'.repeat(1_000), + remedy: 'r'.repeat(1_000) + } + }) + }) + it('lists every active task in addition to the terminal history limit', async () => { const { database } = await createDatabase() const library = database.createKnowledgeBase({ @@ -779,6 +813,63 @@ describe('KnowledgeDatabase', () => { expect(database.deleteEntity(other.id)).toBe(true) }) + it('finds graph identities beyond the ordinary list limit', async () => { + const { database } = await createDatabase() + const knowledgeBase = database.createKnowledgeBase({ + name: 'Large graph identity', + storageMode: 'reference' + }) + let expectedEntityId = '' + const targetEntityIds: string[] = [] + for (let index = 0; index < 501; index += 1) { + const entity = database.createEntity({ + knowledgeBaseId: knowledgeBase.id, + name: `Entity ${String(index).padStart(3, '0')}`, + type: 'CONCEPT', + locked: false + }) + targetEntityIds.push(entity.id) + if (index === 500) { + expectedEntityId = entity.id + } + } + const source = database.createEntity({ + knowledgeBaseId: knowledgeBase.id, + name: 'Relation Source', + type: 'CONCEPT', + locked: false + }) + let expectedRelationId = '' + for (let index = 0; index < 501; index += 1) { + const relation = database.createRelation({ + knowledgeBaseId: knowledgeBase.id, + sourceEntityId: source.id, + targetEntityId: targetEntityIds[index]!, + type: 'RELATED_TO', + locked: false + }) + if (index === 500) { + expectedRelationId = relation.id + } + } + + expect( + database.findEntityByCanonicalName( + knowledgeBase.id, + 'CONCEPT', + 'Entity 500' + )?.id + ).toBe(expectedEntityId) + expect( + database.findRelationByIdentity( + knowledgeBase.id, + source.id, + targetEntityIds[500]!, + 'RELATED_TO' + )?.id + ).toBe(expectedRelationId) + }) + it('persists controlled ontology updates and marks graph rebuild state', async () => { const { database } = await createDatabase() const library = database.createKnowledgeBase({ @@ -1229,6 +1320,12 @@ describe('KnowledgeDatabase', () => { } ] ) + const concurrentReplacementId = + database.beginDocumentEmbeddingReplacement( + document.id, + 'openai-compatible', + 'embed-v1' + ) expect( database.vectorSearch({ @@ -1244,6 +1341,9 @@ describe('KnowledgeDatabase', () => { 'openai-compatible', 'embed-v1' ) + database.discardDocumentEmbeddingReplacement( + concurrentReplacementId + ) expect( database.vectorSearch({ knowledgeBaseId: knowledgeBase.id, @@ -1254,6 +1354,55 @@ describe('KnowledgeDatabase', () => { ).toBe(item.id) }) + it('rejects a staged embedding replacement after chunk content changes', async () => { + const { database } = await createDatabase() + const knowledgeBase = database.createKnowledgeBase({ + name: 'Stale replacement', + storageMode: 'reference' + }) + const seeded = seedDocument( + database, + knowledgeBase.id, + 'stale-replacement' + ) + const document = + database.getEmbeddingIndexDocument(seeded.documentId)! + const item = document.items[0]! + const replacementId = + database.beginDocumentEmbeddingReplacement( + document.id, + 'openai-compatible', + 'embed-v1' + ) + database.appendDocumentEmbeddingBatch( + replacementId, + document.id, + 'openai-compatible', + 'embed-v1', + [{ + chunkId: item.id, + contentChecksum: item.contentChecksum!, + vector: [1, 0] + }] + ) + database.updateChunk({ + knowledgeBaseId: knowledgeBase.id, + documentId: document.id, + chunkId: item.id, + content: 'newer content' + }) + + expect(() => + database.finishDocumentEmbeddingReplacement( + replacementId, + document.id, + 'openai-compatible', + 'embed-v1' + ) + ).toThrow('content changed') + database.discardDocumentEmbeddingReplacement(replacementId) + }) + it('persists embedding index jobs independently by knowledge base', async () => { const created = await createDatabase() let database = created.database @@ -1442,6 +1591,20 @@ describe('KnowledgeDatabase', () => { 0 ).map((chunk) => chunk.id) ).toEqual(['parent-chunk']) + const graphEntity = database.createEntity({ + knowledgeBaseId: library.id, + name: 'Chunk evidence', + type: 'CONCEPT', + locked: false + }) + const evidence = database.createEvidence({ + knowledgeBaseId: library.id, + entityId: graphEntity.id, + documentId: document.id, + chunkId: 'child-chunk', + quote: 'recallable child text', + source: 'rules' + }) database.updateChunk({ knowledgeBaseId: library.id, @@ -1452,6 +1615,12 @@ describe('KnowledgeDatabase', () => { expect( database.search({ knowledgeBaseId: library.id, query: 'recallable' }) ).toEqual([]) + expect(database.listEvidence(library.id)).toEqual([ + expect.objectContaining({ id: evidence.id }) + ]) + expect( + database.graphSearch(library.id, 'Chunk evidence') + ).toEqual([]) database.updateChunk({ knowledgeBaseId: library.id, documentId: document.id, @@ -1462,6 +1631,7 @@ describe('KnowledgeDatabase', () => { expect( database.search({ knowledgeBaseId: library.id, query: '人工修正' }) ).toHaveLength(1) + expect(database.listEvidence(library.id)).toEqual([]) expect( database.listChunksPage({ knowledgeBaseId: library.id, diff --git a/src/main/knowledge/knowledge-database.ts b/src/main/knowledge/knowledge-database.ts index 8dc4bb6..05f26b2 100644 --- a/src/main/knowledge/knowledge-database.ts +++ b/src/main/knowledge/knowledge-database.ts @@ -45,6 +45,7 @@ import { createCjkSearchText, knowledgeRetrievalTerms } from './retrieval-text' +import { normalizeEntityAlias } from './graph-extractor' import type { Chunk, ChunkEmbeddingInput, @@ -79,7 +80,7 @@ import type { VectorSearchOptions } from './types' -const DATABASE_VERSION = 9 +const DATABASE_VERSION = 10 const MAX_ID_LENGTH = 128 const MAX_NAME_LENGTH = 512 const MAX_LOCATION_LENGTH = 8192 @@ -202,6 +203,17 @@ function optionalString( return requiredString(value, field, maximum, trim) } +function truncatedOptionalString( + value: string | null | undefined, + maximum: number +): string | undefined { + if (value === undefined || value === null || value === '') { + return undefined + } + const normalized = String(value).trim() + return normalized ? normalized.slice(0, maximum) : undefined +} + function boundedInteger( value: number, field: string, @@ -1340,25 +1352,20 @@ export class KnowledgeDatabase { ? undefined : update.message === undefined ? current.message - : optionalString( - update.message, - 'message', - MAX_TASK_TEXT_LENGTH - ) + : truncatedOptionalString(update.message, MAX_TASK_TEXT_LENGTH) const error = update.error === null ? undefined : update.error === undefined ? current.error : { - message: requiredString( - update.error.message, - 'error.message', - MAX_TASK_TEXT_LENGTH - ), - remedy: optionalString( + message: + truncatedOptionalString( + update.error.message, + MAX_TASK_TEXT_LENGTH + ) ?? '任务失败', + remedy: truncatedOptionalString( update.error.remedy, - 'error.remedy', MAX_TASK_TEXT_LENGTH ) } @@ -2089,6 +2096,11 @@ export class KnowledgeDatabase { const database = this.requireDatabase() const now = new Date().toISOString() this.transaction(database, () => { + if (content !== current.chunk.content) { + database + .prepare('DELETE FROM graph_evidence WHERE chunk_id = ?') + .run(current.chunk.id) + } database .prepare( `UPDATE chunks SET content = ?, enabled = ?, manually_edited = 1, @@ -2110,9 +2122,6 @@ export class KnowledgeDatabase { database .prepare('DELETE FROM embedding_index_state WHERE document_id = ?') .run(current.document.id) - database - .prepare('DELETE FROM graph_evidence WHERE chunk_id = ?') - .run(current.chunk.id) }) return this.getChunkForReference( input.knowledgeBaseId, @@ -2335,12 +2344,12 @@ export class KnowledgeDatabase { 'documentId', MAX_ID_LENGTH ) - const normalizedProvider = requiredString( + requiredString( provider, 'provider', MAX_EMBEDDING_PROVIDER_LENGTH ) - const normalizedModel = requiredString( + requiredString( model, 'model', MAX_EMBEDDING_MODEL_LENGTH @@ -2353,16 +2362,6 @@ export class KnowledgeDatabase { ) { throw new Error(`Document not found: ${normalizedDocumentId}`) } - database - .prepare( - `DELETE FROM embedding_rebuild_staging - WHERE document_id = ? AND provider = ? AND model = ?` - ) - .run( - normalizedDocumentId, - normalizedProvider, - normalizedModel - ) return randomUUID() } @@ -2544,6 +2543,40 @@ export class KnowledgeDatabase { ) { throw new Error('Embeddings must cover every current document chunk') } + const stagedRows = database + .prepare( + `SELECT + staged.content_checksum AS staged_checksum, + current.index_content AS current_content + FROM embedding_rebuild_staging staged + LEFT JOIN chunks current + ON current.id = staged.chunk_id + AND current.document_id = staged.document_id + AND current.enabled = 1 + AND current.role <> 'parent' + WHERE staged.replacement_id = ? + AND staged.document_id = ? + AND staged.provider = ? + AND staged.model = ?` + ) + .all( + normalizedReplacementId, + normalizedDocumentId, + normalizedProvider, + normalizedModel + ) + if ( + stagedRows.some((row) => { + const currentContent = asOptionalString(row, 'current_content') + return ( + currentContent === undefined || + asString(row, 'staged_checksum') !== + contentChecksum(currentContent) + ) + }) + ) { + throw new Error('Embedding content changed during replacement') + } const indexHash = createHash('sha256') let dimensions: number | undefined let firstChecksum = true @@ -3270,6 +3303,45 @@ export class KnowledgeDatabase { .map(mapEntity) } + findEntityByCanonicalName( + knowledgeBaseId: string, + type: string, + name: string + ): GraphEntity | undefined { + requiredString(knowledgeBaseId, 'knowledgeBaseId', MAX_ID_LENGTH) + const normalizedType = requiredString(type, 'type', MAX_NAME_LENGTH) + const normalizedName = normalizeEntityAlias(name) + if (!normalizedName) { + return undefined + } + return this.listEntitiesForIdentity(knowledgeBaseId) + .find( + (candidate) => + candidate.type === normalizedType && + ( + normalizeEntityAlias(candidate.name) === normalizedName || + candidate.aliases.some( + (alias) => normalizeEntityAlias(alias) === normalizedName + ) + ) + ) + } + + listEntitiesForIdentity(knowledgeBaseId: string): GraphEntity[] { + const normalizedId = requiredString( + knowledgeBaseId, + 'knowledgeBaseId', + MAX_ID_LENGTH + ) + return this.requireDatabase() + .prepare( + `SELECT * FROM graph_entities WHERE knowledge_base_id = ? + ORDER BY name COLLATE NOCASE ASC, id ASC` + ) + .all(normalizedId) + .map(mapEntity) + } + updateEntity(id: string, input: UpdateGraphEntityInput): GraphEntity { const current = this.requiredEntity(id) const ontology = @@ -3430,6 +3502,32 @@ export class KnowledgeDatabase { .map(mapRelation) } + findRelationByIdentity( + knowledgeBaseId: string, + sourceEntityId: string, + targetEntityId: string, + type: string + ): GraphRelation | undefined { + const row = this.requireDatabase() + .prepare( + `SELECT * FROM graph_relations + WHERE knowledge_base_id = ? AND source_entity_id = ? + AND target_entity_id = ? AND type = ? + ORDER BY created_at ASC, id ASC LIMIT 1` + ) + .get( + requiredString( + knowledgeBaseId, + 'knowledgeBaseId', + MAX_ID_LENGTH + ), + requiredString(sourceEntityId, 'sourceEntityId', MAX_ID_LENGTH), + requiredString(targetEntityId, 'targetEntityId', MAX_ID_LENGTH), + requiredString(type, 'type', MAX_NAME_LENGTH) + ) + return row ? mapRelation(row) : undefined + } + updateRelation( id: string, input: UpdateGraphRelationInput @@ -3626,6 +3724,42 @@ export class KnowledgeDatabase { .map(mapEvidence) } + listGraphSnapshot(knowledgeBaseId: string): { + entities: GraphEntity[] + relations: GraphRelation[] + evidence: Evidence[] + } { + const normalizedId = requiredString( + knowledgeBaseId, + 'knowledgeBaseId', + MAX_ID_LENGTH + ) + const database = this.requireDatabase() + return { + entities: database + .prepare( + `SELECT * FROM graph_entities WHERE knowledge_base_id = ? + ORDER BY name COLLATE NOCASE ASC, id ASC` + ) + .all(normalizedId) + .map(mapEntity), + relations: database + .prepare( + `SELECT * FROM graph_relations WHERE knowledge_base_id = ? + ORDER BY created_at ASC, id ASC` + ) + .all(normalizedId) + .map(mapRelation), + evidence: database + .prepare( + `SELECT * FROM graph_evidence WHERE knowledge_base_id = ? + ORDER BY created_at ASC, id ASC` + ) + .all(normalizedId) + .map(mapEvidence) + } + } + updateEvidence(id: string, input: UpdateEvidenceInput): Evidence { const current = this.requiredEvidence(id) const next = { @@ -3961,6 +4095,8 @@ export class KnowledgeDatabase { AND ev.knowledge_base_id = ge.knowledge_base_id AND ec.knowledge_base_id = ge.knowledge_base_id AND ec.document_id = ev.document_id + AND ec.enabled = 1 + AND ec.role <> 'parent' ) ORDER BY ge.name COLLATE NOCASE ASC, ge.id ASC LIMIT 24 @@ -3988,6 +4124,8 @@ export class KnowledgeDatabase { AND rev.knowledge_base_id = gr.knowledge_base_id AND rc.knowledge_base_id = gr.knowledge_base_id AND rc.document_id = rev.document_id + AND rc.enabled = 1 + AND rc.role <> 'parent' ) AND EXISTS ( SELECT 1 FROM graph_evidence nev @@ -4000,6 +4138,8 @@ export class KnowledgeDatabase { AND nev.knowledge_base_id = gr.knowledge_base_id AND nc.knowledge_base_id = gr.knowledge_base_id AND nc.document_id = nev.document_id + AND nc.enabled = 1 + AND nc.role <> 'parent' ) ), reached(id, depth) AS ( @@ -4042,6 +4182,7 @@ export class KnowledgeDatabase { JOIN knowledge_sources s ON s.id = d.source_id WHERE c.knowledge_base_id = ? AND d.knowledge_base_id = ? AND s.knowledge_base_id = ? + AND c.enabled = 1 AND c.role <> 'parent' GROUP BY c.id ORDER BY MIN(backed_evidence.graph_depth) ASC, c.id ASC LIMIT ?` @@ -4175,6 +4316,14 @@ export class KnowledgeDatabase { ) .run(9, new Date().toISOString()) } + if (currentVersion < 10) { + this.migrateToVersion10(database) + database + .prepare( + 'INSERT INTO schema_migrations (version, applied_at) VALUES (?, ?)' + ) + .run(10, new Date().toISOString()) + } database.exec(`PRAGMA user_version = ${DATABASE_VERSION}`) database.exec('COMMIT') } catch (error) { @@ -4783,6 +4932,21 @@ export class KnowledgeDatabase { `) } + private migrateToVersion10(database: DatabaseSync): void { + database.exec(` + CREATE INDEX IF NOT EXISTS graph_evidence_chunk_idx + ON graph_evidence(knowledge_base_id, chunk_id); + CREATE INDEX IF NOT EXISTS graph_evidence_entity_idx + ON graph_evidence(knowledge_base_id, entity_id); + CREATE INDEX IF NOT EXISTS graph_evidence_relation_idx + ON graph_evidence(knowledge_base_id, relation_id); + CREATE INDEX IF NOT EXISTS graph_relations_source_idx + ON graph_relations(knowledge_base_id, source_entity_id); + CREATE INDEX IF NOT EXISTS graph_relations_target_idx + ON graph_relations(knowledge_base_id, target_entity_id); + `) + } + private normalizeChunks(chunks: ReplaceChunkInput[]): Array<{ id: string ordinal: number diff --git a/src/main/knowledge/knowledge-service.test.ts b/src/main/knowledge/knowledge-service.test.ts index bb52c49..9d8d827 100644 --- a/src/main/knowledge/knowledge-service.test.ts +++ b/src/main/knowledge/knowledge-service.test.ts @@ -433,6 +433,37 @@ describe('KnowledgeService', () => { expect(service.database.getEntity(manual.id)).toBeDefined() }) + it('prunes generated graph records after chunk evidence is removed', async () => { + const { directory, service } = await createService() + const sourcePath = join(directory, 'chunk-graph.md') + await writeFile(sourcePath, '# Disposable Entity', 'utf8') + const library = service.createLibrary({ + name: 'Chunk graph cleanup', + storageMode: 'reference', + graphEnabled: true, + graphStrategy: 'rules' + }) + await service.importPaths(library.id, [sourcePath]) + const snapshot = service.snapshot(library.id) + const document = snapshot.documents[0]! + const generatedEntity = snapshot.entities.find( + (entity) => entity.name === 'Disposable Entity' + )! + const evidence = snapshot.evidence.find( + (item) => item.entityId === generatedEntity.id + )! + + await service.updateChunk({ + knowledgeBaseId: library.id, + documentId: document.id, + chunkId: evidence.chunkId!, + content: 'Replacement text without graph evidence' + }) + + expect(service.database.getEntity(generatedEntity.id)).toBeUndefined() + expect(service.snapshot(library.id).evidence).toEqual([]) + }) + it('fails hybrid reextraction when model extraction fails', async () => { const extractStructured = vi.fn(async () => { throw new Error('模型未返回图谱内容') @@ -633,6 +664,44 @@ describe('KnowledgeService', () => { expect(results[0]?.retrieval.channels).toContain('fts') }) + it('bounds long indexing failures so task persistence does not mask them', async () => { + const directory = await mkdtemp( + join(tmpdir(), 'goodbuddy-knowledge-service-') + ) + temporaryDirectories.push(directory) + const service = new KnowledgeService({ + databasePath: join(directory, 'knowledge.sqlite'), + managedRoot: join(directory, 'managed'), + parseDocument: async () => { + throw new Error('x'.repeat(2_000)) + } + }) + await service.initialize() + services.push(service) + const sourcePath = join(directory, 'failure.txt') + await writeFile(sourcePath, 'failure source', 'utf8') + const library = service.createLibrary({ + name: 'Bounded task errors', + storageMode: 'reference', + graphEnabled: false + }) + + await expect( + service.importPaths(library.id, [sourcePath]) + ).rejects.toThrow() + const failedTasks = service + .snapshot(library.id) + .tasks.filter((task) => task.status === 'failed') + expect(failedTasks.length).toBeGreaterThan(0) + expect( + failedTasks.every( + (task) => + (task.message?.length ?? 0) <= 1_000 && + (task.error?.message.length ?? 0) <= 1_000 + ) + ).toBe(true) + }) + it('defers existing-document rebuilds when an embedding provider is enabled', async () => { const { directory, service } = await createService() const sourcePath = join(directory, 'existing.txt') @@ -1202,6 +1271,149 @@ describe('KnowledgeService', () => { }) }) + it('marks an existing URL source failed when refresh fails before parsing', async () => { + const importer = { + import: vi.fn(async () => { + throw new Error('URL refresh unavailable') + }) + } as unknown as UrlImporter + const { service } = await createService(importer) + const library = service.createLibrary({ + name: 'Failed URL refresh', + storageMode: 'managed', + graphEnabled: false + }) + const source = service.database.upsertSource({ + knowledgeBaseId: library.id, + type: 'url', + location: 'https://example.com/failure', + displayName: 'Existing URL', + status: 'ready' + }) + + await expect(service.syncSource(source.id)).rejects.toThrow( + 'URL refresh unavailable' + ) + expect(service.database.getSource(source.id)).toMatchObject({ + status: 'error', + lastError: 'URL refresh unavailable' + }) + }) + + it('preserves refreshed URL metadata when later indexing fails', async () => { + const importer = { + import: vi.fn(async () => ({ + url: 'https://example.com/redirected', + title: 'Redirected title', + contentType: 'text/html', + etag: 'fresh-etag', + lastModified: undefined, + discoveredUrls: [], + document: { + title: 'Redirected title', + sourceFormat: '.html', + content: 'refreshed content', + sections: [{ + locator: '网页正文', + content: 'refreshed content' + }], + warnings: [] + } + })) + } as unknown as UrlImporter + const { service } = await createService(importer) + const library = service.createLibrary({ + name: 'Failed refreshed URL', + storageMode: 'managed', + graphEnabled: false + }) + const source = service.database.upsertSource({ + knowledgeBaseId: library.id, + type: 'url', + location: 'https://example.com/original', + displayName: 'Original title', + status: 'ready' + }) + vi.spyOn(service.database, 'upsertDocument').mockImplementationOnce( + () => { + throw new Error('synthetic indexing failure') + } + ) + + await expect(service.syncSource(source.id)).rejects.toThrow( + 'synthetic indexing failure' + ) + expect(service.database.getSource(source.id)).toMatchObject({ + location: 'https://example.com/redirected', + displayName: 'Redirected title', + status: 'error', + metadata: expect.objectContaining({ etag: 'fresh-etag' }) + }) + }) + + it('serializes background embedding reindexes and awaits the rerun on disposal', async () => { + const resolvers: Array<() => void> = [] + let activeEmbeddings = 0 + let maximumActiveEmbeddings = 0 + const provider: EmbeddingProvider = { + provider: 'serial-background-provider', + model: 'serial-background-model', + embed: vi.fn(async () => { + activeEmbeddings += 1 + maximumActiveEmbeddings = Math.max( + maximumActiveEmbeddings, + activeEmbeddings + ) + await new Promise((resolve) => { + resolvers.push(resolve) + }) + activeEmbeddings -= 1 + return [[1, 0]] + }) + } + const { directory, service } = await createService() + const sourcePath = join(directory, 'serial-background.txt') + await writeFile(sourcePath, 'initial content', 'utf8') + const library = service.createLibrary({ + name: 'Serialized background embeddings', + storageMode: 'reference', + graphEnabled: false + }) + await service.importPaths(library.id, [sourcePath]) + await service.setEmbeddingProvider(provider) + const document = service.snapshot(library.id).documents[0]! + const chunk = service.database.listChunks(document.id, 1)[0]! + + await service.updateChunk({ + knowledgeBaseId: library.id, + documentId: document.id, + chunkId: chunk.id, + content: 'first edit' + }) + await vi.waitFor(() => expect(resolvers).toHaveLength(1)) + await service.updateChunk({ + knowledgeBaseId: library.id, + documentId: document.id, + chunkId: chunk.id, + content: 'second edit' + }) + await new Promise((resolve) => setTimeout(resolve, 300)) + expect(maximumActiveEmbeddings).toBe(1) + + resolvers.shift()?.() + await vi.waitFor(() => expect(resolvers).toHaveLength(1)) + const disposing = service.dispose() + let disposed = false + void disposing.then(() => { + disposed = true + }) + await new Promise((resolve) => setTimeout(resolve, 0)) + expect(disposed).toBe(false) + resolvers.shift()?.() + await disposing + expect(maximumActiveEmbeddings).toBe(1) + }) + it('cancels and awaits active graph work before deleting a library', async () => { let extractionSignal: AbortSignal | undefined const extractStructured: ExtractStructured = (_prompt, signal) => { @@ -1280,6 +1492,54 @@ describe('KnowledgeService', () => { ).toBe(true) }) + it('deduplicates a source retry against an active manual sync', async () => { + let releaseImport: (() => void) | undefined + const importer = { + import: vi.fn( + async () => { + await new Promise((resolve) => { + releaseImport = resolve + }) + throw new Error('synthetic sync failure') + } + ) + } as unknown as UrlImporter + const { service } = await createService(importer) + const library = service.createLibrary({ + name: 'Dedupe source retry', + storageMode: 'managed', + graphEnabled: false + }) + const source = service.database.upsertSource({ + knowledgeBaseId: library.id, + type: 'url', + location: 'https://example.com/dedupe', + displayName: 'Dedupe URL', + status: 'ready' + }) + const failed = service.database.createKnowledgeTask({ + libraryId: library.id, + sourceId: source.id, + documentName: source.displayName, + scope: 'source', + kind: 'source-sync', + status: 'failed', + error: { message: 'retry me' } + }) + + const syncing = service.syncSource(source.id) + void syncing.catch(() => undefined) + await vi.waitFor(() => expect(importer.import).toHaveBeenCalledOnce()) + const retrying = service.retryTask(failed.id) + void retrying.catch(() => undefined) + await new Promise((resolve) => setTimeout(resolve, 0)) + expect(importer.import).toHaveBeenCalledOnce() + + releaseImport?.() + await expect(syncing).rejects.toThrow('synthetic sync failure') + await expect(retrying).rejects.toThrow('synthetic sync failure') + }) + it('reconciles embedding task status during ordinary snapshots', async () => { let resolveEmbedding: ((value: number[][]) => void) | undefined const provider: EmbeddingProvider = { diff --git a/src/main/knowledge/knowledge-service.ts b/src/main/knowledge/knowledge-service.ts index 27e1bf3..d0e2c1c 100644 --- a/src/main/knowledge/knowledge-service.ts +++ b/src/main/knowledge/knowledge-service.ts @@ -2,7 +2,6 @@ import { cp, lstat, mkdir, - open, readdir, realpath, rm, @@ -91,7 +90,10 @@ import type { } from './types' import { UrlImporter } from './url-importer' import { mimeTypeFromFileName } from '../file-media-type' -import { isPathInside } from '../workspace-file-access' +import { + isPathInside, + readBoundedFile +} from '../workspace-file-access' type ScannedFile = { absolutePath: string @@ -182,6 +184,11 @@ export class KnowledgeService { string, Promise >() + private readonly backgroundEmbeddingReindexes = new Map< + string, + Promise + >() + private readonly pendingEmbeddingReindexes = new Set() private readonly taskControllers = new Map() private readonly taskOperations = new Map>() private readonly sourceSyncTaskIds = new Map() @@ -253,6 +260,7 @@ export class KnowledgeService { clearTimeout(timer) } this.embeddingEditTimers.clear() + this.pendingEmbeddingReindexes.clear() for (const watcher of this.watchers.values()) { watcher.close() } @@ -266,6 +274,7 @@ export class KnowledgeService { await Promise.allSettled([ ...this.activeSyncs.values(), ...this.activeDocumentRebuilds.values(), + ...this.backgroundEmbeddingReindexes.values(), ...this.taskOperations.values(), ...embeddingCompletions ]) @@ -451,6 +460,11 @@ export class KnowledgeService { }) } + private static taskErrorMessage(error: unknown): string { + const message = error instanceof Error ? error.message : '任务失败' + return message.slice(0, 1_000) || '任务失败' + } + private failKnowledgeTask(taskId: string, error: unknown): void { const current = this.database.getKnowledgeTask(taskId) if ( @@ -459,7 +473,7 @@ export class KnowledgeService { ) { return } - const message = error instanceof Error ? error.message : '任务失败' + const message = KnowledgeService.taskErrorMessage(error) this.database.updateKnowledgeTask(taskId, { status: 'failed', message, @@ -712,13 +726,14 @@ export class KnowledgeService { : undefined } }) + const graph = this.database.listGraphSnapshot(libraryId) return { libraries, sources, documents, - entities: this.database.listEntities(libraryId), - relations: this.database.listRelations(libraryId), - evidence: this.database.listEvidence(libraryId), + entities: graph.entities, + relations: graph.relations, + evidence: graph.evidence, tasks: this.database.listKnowledgeTasks(libraryId) } } @@ -1223,6 +1238,15 @@ export class KnowledgeService { ) { this.scheduleEmbeddingReindex(input.documentId) } + if ( + current && + input.content !== undefined && + input.content !== current.content + ) { + this.database.pruneUnreferencedGeneratedGraph( + input.knowledgeBaseId + ) + } return chunk } @@ -1231,6 +1255,9 @@ export class KnowledgeService { const deleted = this.database.deleteChunk(input) if (deleted) { this.scheduleEmbeddingReindex(input.documentId) + this.database.pruneUnreferencedGeneratedGraph( + input.knowledgeBaseId + ) } return deleted } @@ -1618,6 +1645,7 @@ export class KnowledgeService { effectiveLibrary ) ) + this.database.pruneUnreferencedGeneratedGraph(library.id) this.updateKnowledgeTask(parsingTask.id, { documentId: document.id, documentName: document.title, @@ -1702,11 +1730,21 @@ export class KnowledgeService { } async syncSource(sourceId: string): Promise { + return this.syncSourceOperation(sourceId) + } + + private async syncSourceOperation( + sourceId: string, + retryOfTaskId?: string + ): Promise { const existing = this.activeSyncs.get(sourceId) if (existing) { return existing } - const operation = this.performSyncSource(sourceId).finally(() => { + const operation = this.performSyncSource( + sourceId, + retryOfTaskId + ).finally(() => { if (this.activeSyncs.get(sourceId) === operation) { this.activeSyncs.delete(sourceId) } @@ -1836,7 +1874,12 @@ export class KnowledgeService { progress: 5, message: '正在读取文档' }) - const buffer = await this.readBoundedFile(location) + const buffer = await readBoundedFile( + location, + maximumFileBytes, + '文件超过 20MB', + '文件不是普通文件' + ) effectiveSignal.throwIfAborted() this.updateKnowledgeTask(task.id, { stage: 'parsing', @@ -1868,6 +1911,7 @@ export class KnowledgeService { }, this.createDocumentChunks(parsed, library) ) + this.database.pruneUnreferencedGeneratedGraph(library.id) this.updateKnowledgeTask(task.id, { stage: 'embedding', progress: 65, @@ -2073,7 +2117,7 @@ export class KnowledgeService { if (!task.sourceId) { throw new Error('任务缺少知识来源') } - await this.performSyncSource(task.sourceId, task.id) + await this.syncSourceOperation(task.sourceId, task.id) return } case 'document-rebuild': { @@ -2151,7 +2195,8 @@ export class KnowledgeService { if (library.graphStrategy === 'ask') { throw new Error('按需询问策略不会自动抽取，请在设置中选择其他策略') } - const documents = this.database.listDocuments(library.id) + const documents = + this.database.listDocumentsForLibraryRebuild(library.id) const parentTask = this.createKnowledgeTask({ libraryId: library.id, retryOfTaskId, @@ -2293,6 +2338,9 @@ export class KnowledgeService { 'Knowledge source deleted' ) const removed = this.database.removeSource(sourceId) + if (removed) { + this.database.pruneUnreferencedGeneratedGraph(library.id) + } if ( removed && library.storageMode === 'managed' && @@ -2395,13 +2443,18 @@ export class KnowledgeService { } else { this.failKnowledgeTask(task.id, error) } - if (!effectiveSignal.aborted && source.type !== 'url') { - this.database.upsertSource({ - ...source, - status: 'error', - lastError: - error instanceof Error ? error.message.slice(0, 1_000) : '同步失败' - }) + if (!effectiveSignal.aborted) { + const currentSource = this.database.getSource(source.id) + if (currentSource) { + this.database.upsertSource({ + ...currentSource, + status: 'error', + lastError: + error instanceof Error + ? error.message.slice(0, 1_000) + : '同步失败' + }) + } } throw error } finally { @@ -2431,6 +2484,11 @@ export class KnowledgeService { this.database.removeDocument(document.id) } } + if (existing.some( + (document) => !currentExternalIds.has(document.externalId) + )) { + this.database.pruneUnreferencedGeneratedGraph(library.id) + } const failures: string[] = [] for (let index = 0; index < files.length; index += 1) { @@ -2454,7 +2512,12 @@ export class KnowledgeService { progress: 10, message: '正在读取文档' }) - const buffer = await this.readBoundedFile(file.absolutePath) + const buffer = await readBoundedFile( + file.absolutePath, + maximumFileBytes, + '文件超过 20MB', + '文件不是普通文件' + ) signal.throwIfAborted() this.updateKnowledgeTask(parsingTask.id, { stage: 'parsing', @@ -2507,6 +2570,7 @@ export class KnowledgeService { }, this.createDocumentChunks(parsed, effectiveLibrary) ) + this.database.pruneUnreferencedGeneratedGraph(library.id) this.updateKnowledgeTask(parsingTask.id, { documentId: document.id, documentName: document.title, @@ -2550,7 +2614,7 @@ export class KnowledgeService { this.failKnowledgeTask(parsingTask.id, error) failures.push( `${file.relativePath}: ${ - error instanceof Error ? error.message : '解析失败' + KnowledgeService.taskErrorMessage(error) }` ) } @@ -2564,8 +2628,9 @@ export class KnowledgeService { }) } if (failures.length > 0) { + const detail = failures.slice(0, 5).join('；') throw new Error( - `${failures.length} 个文件处理失败：${failures.slice(0, 5).join('；')}` + `${failures.length} 个文件处理失败：${detail}`.slice(0, 1_000) ) } } @@ -2577,15 +2642,49 @@ export class KnowledgeService { } const timer = setTimeout(() => { this.embeddingEditTimers.delete(documentId) - const document = this.database.getDocument(documentId) - if (document && this.embeddingProvider) { - void this.indexDocumentEmbeddings(document) + if (this.backgroundEmbeddingReindexes.has(documentId)) { + this.pendingEmbeddingReindexes.add(documentId) + return } + this.startBackgroundEmbeddingReindex(documentId) }, 250) timer.unref?.() this.embeddingEditTimers.set(documentId, timer) } + private startBackgroundEmbeddingReindex(documentId: string): void { + if ( + this.backgroundEmbeddingReindexes.has(documentId) || + this.lifecycleController.signal.aborted + ) { + return + } + const operation = (async () => { + do { + this.pendingEmbeddingReindexes.delete(documentId) + const document = this.database.getDocument(documentId) + if ( + !document || + !this.embeddingProvider || + this.lifecycleController.signal.aborted + ) { + return + } + await this.indexDocumentEmbeddings(document) + } while (this.pendingEmbeddingReindexes.delete(documentId)) + })() + .catch(() => undefined) + .finally(() => { + this.pendingEmbeddingReindexes.delete(documentId) + if ( + this.backgroundEmbeddingReindexes.get(documentId) === operation + ) { + this.backgroundEmbeddingReindexes.delete(documentId) + } + }) + this.backgroundEmbeddingReindexes.set(documentId, operation) + } + private cancelScheduledEmbeddingReindex(documentId: string): void { const timer = this.embeddingEditTimers.get(documentId) if (timer) { @@ -2850,23 +2949,24 @@ export class KnowledgeService { document: Document, result: GraphExtractionResult ): void { - const existingEntities = this.database.listEntities(library.id) const chunksById = new Map( this.database .listChunks(document.id) .map((chunk) => [chunk.id, chunk]) ) const entityIds = new Map() + const existingEntitiesByIdentity = new Map() + for (const entity of this.database.listEntitiesForIdentity(library.id)) { + for (const name of [entity.name, ...entity.aliases]) { + existingEntitiesByIdentity.set( + `${entity.type}\0${normalizeEntityAlias(name)}`, + entity + ) + } + } for (const entity of result.entities) { - const normalized = normalizeEntityAlias(entity.name) - const existing = existingEntities.find( - (candidate) => - candidate.type === entity.type && - (normalizeEntityAlias(candidate.name) === normalized || - candidate.aliases.some( - (alias) => normalizeEntityAlias(alias) === normalized - )) - ) + const identity = `${entity.type}\0${normalizeEntityAlias(entity.name)}` + const existing = existingEntitiesByIdentity.get(identity) const stored = existing ? existing.locked ? existing @@ -2880,6 +2980,12 @@ export class KnowledgeService { aliases: entity.aliases, locked: false }) + for (const name of [stored.name, ...stored.aliases]) { + existingEntitiesByIdentity.set( + `${stored.type}\0${normalizeEntityAlias(name)}`, + stored + ) + } entityIds.set(entity.id, stored.id) for (const evidence of entity.evidence) { this.database.createEvidence({ @@ -2900,18 +3006,17 @@ export class KnowledgeService { }) } } - const existingRelations = this.database.listRelations(library.id) for (const relation of result.relations) { const sourceEntityId = entityIds.get(relation.sourceId) const targetEntityId = entityIds.get(relation.targetId) if (!sourceEntityId || !targetEntityId) { continue } - const existing = existingRelations.find( - (candidate) => - candidate.sourceEntityId === sourceEntityId && - candidate.targetEntityId === targetEntityId && - candidate.type === relation.type + const existing = this.database.findRelationByIdentity( + library.id, + sourceEntityId, + targetEntityId, + relation.type ) const stored = existing ?? @@ -3027,24 +3132,6 @@ export class KnowledgeService { } } - private async readBoundedFile(path: string): Promise { - const handle = await open(path, 'r') - try { - const fileStat = await handle.stat() - if (!fileStat.isFile() || fileStat.size > maximumFileBytes) { - throw new Error('文件超过 20MB 或不是普通文件') - } - const buffer = Buffer.alloc(fileStat.size + 1) - const result = await handle.read(buffer, 0, buffer.length, 0) - if (result.bytesRead > maximumFileBytes) { - throw new Error('文件超过 20MB') - } - return buffer.subarray(0, result.bytesRead) - } finally { - await handle.close() - } - } - private startWatcher(source: KnowledgeSource): void { this.stopWatcher(source.id) try {