fix: preserve knowledge indexing state

This commit is contained in:
lofyer
2026-08-13 04:47:44 +08:00
parent e3b5702767
commit 0c46afba59
4 changed files with 772 additions and 91 deletions
+176 -6
View File
@@ -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,
+192 -28
View File
@@ -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
@@ -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<void>((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<void>((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 = {
+144 -57
View File
@@ -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<Document>
>()
private readonly backgroundEmbeddingReindexes = new Map<
string,
Promise<void>
>()
private readonly pendingEmbeddingReindexes = new Set<string>()
private readonly taskControllers = new Map<string, AbortController>()
private readonly taskOperations = new Map<string, Promise<unknown>>()
private readonly sourceSyncTaskIds = new Map<string, string>()
@@ -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<void> {
return this.syncSourceOperation(sourceId)
}
private async syncSourceOperation(
sourceId: string,
retryOfTaskId?: string
): Promise<void> {
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<string, string>()
const existingEntitiesByIdentity = new Map<string, GraphEntity>()
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<Buffer> {
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 {