chore: prepare GoodBuddy 0.8.2
Cross-platform packages / Validate source (push) Waiting to run
Cross-platform packages / ${{ matrix.platform }} ${{ matrix.arch }} (arm64, macos, macos-15) (push) Blocked by required conditions
Cross-platform packages / ${{ matrix.platform }} ${{ matrix.arch }} (arm64, windows, windows-2025) (push) Blocked by required conditions
Cross-platform packages / ${{ matrix.platform }} ${{ matrix.arch }} (x64, linux, ubuntu-24.04) (push) Blocked by required conditions
Cross-platform packages / ${{ matrix.platform }} ${{ matrix.arch }} (x64, macos, macos-15-intel) (push) Blocked by required conditions
Cross-platform packages / ${{ matrix.platform }} ${{ matrix.arch }} (x64, windows, windows-2025) (push) Blocked by required conditions
Cross-platform packages / Publish GitHub Release (push) Blocked by required conditions
Cross-platform packages / ${{ matrix.platform }} ${{ matrix.arch }} (arm64, linux, ubuntu-24.04-arm) (push) Blocked by required conditions

This commit is contained in:
lofyer
2026-08-06 22:47:13 +08:00
parent 8d00e6371d
commit b8fc7bc86e
114 changed files with 22916 additions and 1560 deletions
@@ -0,0 +1,75 @@
import { describe, expect, it } from 'vitest'
import {
classifyEmbeddingError,
EmbeddingOperationError,
toEmbeddingOperationError
} from './embedding-errors'
describe('embedding error classification', () => {
it.each([
[new Error('Embedding request failed with HTTP 404'), 'model_not_found'],
[new Error('unknown model vendor/embed-v9'), 'model_not_found'],
[{ status: 401 }, 'authentication'],
[new Error('Incorrect API key provided'), 'authentication'],
[{ statusCode: 429 }, 'rate_limited'],
[new Error('request ETIMEDOUT'), 'timeout'],
[new TypeError('fetch failed'), 'network'],
[{ code: 503 }, 'provider_unavailable']
])('classifies %p as %s', (error, code) => {
expect(classifyEmbeddingError(error).code).toBe(code)
})
it('distinguishes explicit cancellation from timeout aborts', () => {
const abort = new Error('The operation was aborted')
abort.name = 'AbortError'
expect(classifyEmbeddingError(abort).code).toBe('cancelled')
expect(
classifyEmbeddingError(abort, { timedOut: true }).code
).toBe('cancelled')
expect(
classifyEmbeddingError(new Error('request stopped'), {
timedOut: true
}).code
).toBe('timeout')
})
it('never returns provider bodies, credentials, endpoints or nested causes', () => {
const secret =
'sk-secret-value https://vectors.example/v1 {"private":"document"}'
const source = Object.assign(new Error(secret), {
status: 401,
response: {
body: secret,
headers: { authorization: `Bearer ${secret}` }
},
cause: new Error(secret)
})
const result = classifyEmbeddingError(source)
const serialized = JSON.stringify(result)
expect(result).toEqual({
code: 'authentication',
message: '向量服务身份验证失败。',
retryable: false,
remedy: '请检查访问密钥是否有效以及是否具备调用向量模型的权限。'
})
expect(serialized).not.toContain('secret')
expect(serialized).not.toContain('vectors.example')
expect(serialized).not.toContain('private')
})
it('wraps unknown errors in a safe serializable operation error', () => {
const wrapped = toEmbeddingOperationError(
new Error('raw provider payload with token')
)
expect(wrapped).toBeInstanceOf(EmbeddingOperationError)
expect(wrapped.toSafeError()).toEqual({
code: 'unknown',
message: '向量操作失败。',
retryable: false,
remedy: '请检查向量服务配置后重试。'
})
expect(JSON.stringify(wrapped.toSafeError())).not.toContain('token')
expect(toEmbeddingOperationError(wrapped)).toBe(wrapped)
})
})
+255
View File
@@ -0,0 +1,255 @@
import type {
EmbeddingErrorCode,
EmbeddingSafeError
} from '../../shared/embedding-contracts'
const MAX_SAFE_MESSAGE_LENGTH = 500
const descriptors: Record<
EmbeddingErrorCode,
Omit<EmbeddingSafeError, 'code'>
> = {
model_not_found: {
message: '未找到指定的向量模型。',
retryable: false,
remedy: '请确认模型名称正确,并确认该模型已在服务端启用。'
},
authentication: {
message: '向量服务身份验证失败。',
retryable: false,
remedy: '请检查访问密钥是否有效以及是否具备调用向量模型的权限。'
},
rate_limited: {
message: '向量服务当前请求过多。',
retryable: true,
remedy: '请稍后重试,或检查服务配额与速率限制。'
},
timeout: {
message: '向量服务响应超时。',
retryable: true,
remedy: '请检查网络和服务状态,然后重试。'
},
network: {
message: '无法连接到向量服务。',
retryable: true,
remedy: '请检查服务地址、网络连接和代理设置。'
},
provider_unavailable: {
message: '向量服务暂时不可用。',
retryable: true,
remedy: '请稍后重试并检查服务运行状态。'
},
invalid_configuration: {
message: '向量模型配置无效。',
retryable: false,
remedy: '请检查服务地址、模型名称和配置参数。'
},
invalid_response: {
message: '向量服务返回了无效结果。',
retryable: false,
remedy: '请确认服务兼容 OpenAI 向量接口并返回有效向量。'
},
cancelled: {
message: '向量操作已取消。',
retryable: true
},
unknown: {
message: '向量操作失败。',
retryable: false,
remedy: '请检查向量服务配置后重试。'
}
}
function errorText(error: unknown): string {
if (error instanceof Error) {
return `${error.name} ${error.message}`.toLowerCase()
}
return typeof error === 'string' ? error.toLowerCase() : ''
}
function numericStatus(error: unknown): number | undefined {
if (typeof error !== 'object' || error === null) {
return undefined
}
for (const key of ['status', 'statusCode', 'code'] as const) {
const value = Reflect.get(error, key)
if (typeof value === 'number' && Number.isInteger(value)) {
return value
}
if (
typeof value === 'string' &&
/^\d{3}$/u.test(value) &&
Number.isInteger(Number(value))
) {
return Number(value)
}
}
return undefined
}
function statusFromText(text: string): number | undefined {
const match = /\b(?:http|status(?: code)?)\s*[:=]?\s*(\d{3})\b/iu.exec(
text
)
return match?.[1] ? Number(match[1]) : undefined
}
function hasAny(text: string, patterns: readonly string[]): boolean {
return patterns.some((pattern) => text.includes(pattern))
}
function classifyCode(
error: unknown,
options: { cancelled?: boolean; timedOut?: boolean }
): EmbeddingErrorCode {
const text = errorText(error)
const status = numericStatus(error) ?? statusFromText(text)
if (
options.cancelled ||
hasAny(text, ['aborterror', 'aborted', 'cancelled', 'canceled'])
) {
return 'cancelled'
}
if (
options.timedOut ||
hasAny(text, ['timeout', 'timed out', 'etimedout'])
) {
return 'timeout'
}
if (
status === 401 ||
status === 403 ||
hasAny(text, [
'unauthorized',
'forbidden',
'authentication',
'invalid api key',
'incorrect api key'
])
) {
return 'authentication'
}
if (
status === 404 ||
hasAny(text, [
'model not found',
'model_not_found',
'unknown model',
'does not exist'
])
) {
return 'model_not_found'
}
if (
status === 429 ||
hasAny(text, ['rate limit', 'rate_limit', 'too many requests', 'quota'])
) {
return 'rate_limited'
}
if (
status === 408 ||
status === 504
) {
return 'timeout'
}
if (
status !== undefined &&
status >= 500 &&
status <= 599
) {
return 'provider_unavailable'
}
if (
hasAny(text, [
'econnrefused',
'econnreset',
'enotfound',
'fetch failed',
'network',
'failed to fetch',
'socket'
])
) {
return 'network'
}
if (
error instanceof RangeError ||
hasAny(text, [
'endpoint must',
'model must',
'invalid endpoint',
'invalid configuration'
])
) {
return 'invalid_configuration'
}
if (
error instanceof TypeError ||
hasAny(text, [
'invalid dimensions',
'invalid result',
'invalid indexes',
'inconsistent dimensions',
'finite numbers',
'valid json',
'invalid response'
])
) {
return 'invalid_response'
}
return 'unknown'
}
/**
* Converts provider and transport failures into a bounded, user-safe error.
* Raw provider response bodies, endpoints, keys and nested causes are never
* copied into the returned value.
*/
export function classifyEmbeddingError(
error: unknown,
options: { cancelled?: boolean; timedOut?: boolean } = {}
): EmbeddingSafeError {
const code = classifyCode(error, options)
const descriptor = descriptors[code]
return {
code,
message: descriptor.message.slice(0, MAX_SAFE_MESSAGE_LENGTH),
retryable: descriptor.retryable,
...(descriptor.remedy
? { remedy: descriptor.remedy.slice(0, MAX_SAFE_MESSAGE_LENGTH) }
: {})
}
}
export class EmbeddingOperationError extends Error {
readonly code: EmbeddingErrorCode
readonly retryable: boolean
readonly remedy?: string
constructor(error: EmbeddingSafeError) {
super(error.message)
this.name = 'EmbeddingOperationError'
this.code = error.code
this.retryable = error.retryable
this.remedy = error.remedy
}
toSafeError(): EmbeddingSafeError {
return {
code: this.code,
message: this.message,
retryable: this.retryable,
...(this.remedy ? { remedy: this.remedy } : {})
}
}
}
export function toEmbeddingOperationError(
error: unknown,
options?: { cancelled?: boolean; timedOut?: boolean }
): EmbeddingOperationError {
return error instanceof EmbeddingOperationError
? error
: new EmbeddingOperationError(classifyEmbeddingError(error, options))
}
@@ -0,0 +1,370 @@
import { describe, expect, it, vi } from 'vitest'
import type {
EmbeddingIndexJob,
EmbeddingIndexStatus
} from '../../shared/embedding-contracts'
import {
EmbeddingIndexCoordinator,
type EmbeddingIndexProvider,
type EmbeddingIndexRecord,
type EmbeddingIndexRepository
} from './embedding-index-coordinator'
class MemoryRepository implements EmbeddingIndexRepository {
readonly documents = [
{
id: 'document-1',
items: [
{ id: 'chunk-1', content: 'alpha', contentChecksum: 'sum-1' }
]
},
{
id: 'document-2',
items: [
{ id: 'chunk-2', content: 'beta', contentChecksum: 'sum-2' },
{ id: 'chunk-3', content: 'gamma', contentChecksum: 'sum-3' }
]
}
]
readonly records = new Map<string, readonly EmbeddingIndexRecord[]>([
[
'document-1',
[{ itemId: 'chunk-1', contentChecksum: 'sum-1', vector: [0, 1] }]
],
[
'document-2',
[
{ itemId: 'chunk-2', contentChecksum: 'sum-2', vector: [0, 1] },
{ itemId: 'chunk-3', contentChecksum: 'sum-3', vector: [0, 1] }
]
]
])
readonly errors = new Map<string, string>()
readonly pendingRecords = new Map<
string,
{ documentId: string; records: EmbeddingIndexRecord[] }
>()
readonly events: string[] = []
lastJob: EmbeddingIndexJob | null = null
readonly savedStatuses: EmbeddingIndexStatus[] = []
async getLastJob(): Promise<EmbeddingIndexJob | null> {
return this.lastJob
}
async saveStatus(status: EmbeddingIndexStatus): Promise<void> {
this.savedStatuses.push(status)
this.lastJob = status.job
}
async listIndexDocumentIds(): Promise<string[]> {
return this.documents.map((document) => document.id)
}
async getIndexDocument(
documentId: string
): Promise<(typeof this.documents)[number] | undefined> {
this.events.push(`load:${documentId}`)
return this.documents.find((document) => document.id === documentId)
}
async beginDocumentReplacement(
documentId: string
): Promise<string> {
const replacementId = `replacement-${documentId}`
this.pendingRecords.set(replacementId, {
documentId,
records: []
})
this.events.push(`begin:${documentId}`)
return replacementId
}
async appendDocumentReplacement(
replacementId: string,
_documentId: string,
_provider: string,
_model: string,
records: readonly EmbeddingIndexRecord[]
): Promise<void> {
this.pendingRecords.get(replacementId)?.records.push(...records)
this.events.push(`append:${replacementId}`)
}
async finishDocumentReplacement(
replacementId: string,
documentId: string
): Promise<void> {
const pending = this.pendingRecords.get(replacementId)
if (!pending) {
throw new Error('Missing pending replacement')
}
this.records.set(documentId, pending.records)
this.errors.delete(documentId)
this.pendingRecords.delete(replacementId)
this.events.push(`finish:${documentId}`)
}
async discardDocumentReplacement(
replacementId: string
): Promise<void> {
this.pendingRecords.delete(replacementId)
this.events.push(`discard:${replacementId}`)
}
async recordDocumentError(
documentId: string,
_provider: string,
_model: string,
error: string
): Promise<void> {
this.errors.set(documentId, error)
}
}
function provider(
embed: EmbeddingIndexProvider['embed']
): EmbeddingIndexProvider {
return {
provider: 'openai-compatible',
model: 'embed-v2',
fingerprint: 'openai-compatible:https://safe.invalid:embed-v2',
embed
}
}
describe('EmbeddingIndexCoordinator', () => {
it('performs a real embedding request for diagnostics', async () => {
const repository = new MemoryRepository()
const embed = vi.fn(async () => [[0.25, 0.5, 0.75]])
const times = [100, 137]
const coordinator = new EmbeddingIndexCoordinator(repository, {
now: () => times.shift() ?? 137
})
await expect(coordinator.diagnose(provider(embed))).resolves.toEqual({
status: 'available',
provider: 'openai-compatible',
model: 'embed-v2',
checkedAt: 137,
latencyMs: 37,
dimensions: 3
})
expect(embed).toHaveBeenCalledWith(
['GoodBuddy 向量模型连接测试'],
undefined
)
})
it('reports a safe diagnostic failure instead of treating config as success', async () => {
const repository = new MemoryRepository()
const coordinator = new EmbeddingIndexCoordinator(repository, {
now: () => 100
})
const result = await coordinator.diagnose(
provider(async () => {
throw Object.assign(
new Error('Bearer sk-secret failed with private payload'),
{ status: 401 }
)
})
)
expect(result).toMatchObject({
status: 'unavailable',
error: {
code: 'authentication',
retryable: false
}
})
expect(JSON.stringify(result)).not.toContain('sk-secret')
expect(JSON.stringify(result)).not.toContain('private payload')
})
it('replaces each document atomically and persists completed progress', async () => {
const repository = new MemoryRepository()
const coordinator = new EmbeddingIndexCoordinator(repository, {
batchSize: 2,
now: (() => {
let value = 10
return () => value++
})(),
createId: () => 'job-1'
})
await coordinator.initialize()
const statuses: string[] = []
coordinator.subscribe((status) => {
statuses.push(status.job?.status ?? 'idle')
})
const queued = coordinator.startRebuild(
provider(async (input) =>
input.map((text) =>
text === 'beta' ? [0, 1] : [1, 0]
)
)
)
expect(queued.status).toBe('queued')
const completed = await coordinator.waitForCompletion()
expect(completed).toMatchObject({
status: 'completed',
progress: { completed: 2, total: 2, percent: 100 }
})
expect(statuses).toContain('queued')
expect(statuses).toContain('running')
expect(statuses.at(-1)).toBe('completed')
expect(repository.records.get('document-1')).toEqual([
expect.objectContaining({ itemId: 'chunk-1', vector: [1, 0] })
])
expect(repository.records.get('document-2')).toEqual([
expect.objectContaining({ itemId: 'chunk-2', vector: [0, 1] }),
expect.objectContaining({ itemId: 'chunk-3', vector: [1, 0] })
])
expect(repository.events.indexOf('finish:document-1')).toBeLessThan(
repository.events.indexOf('load:document-2')
)
expect(repository.pendingRecords.size).toBe(0)
expect(repository.lastJob).toEqual(completed)
})
it('keeps completed documents and records a safe error on the failed document', async () => {
const repository = new MemoryRepository()
const coordinator = new EmbeddingIndexCoordinator(repository, {
batchSize: 2,
now: () => 20,
createId: () => 'job-failed'
})
await coordinator.initialize()
coordinator.startRebuild(
provider(async (input) => {
if (input.includes('beta')) {
throw Object.assign(new Error('raw upstream token sk-private'), {
status: 429
})
}
return input.map(() => [1, 0])
})
)
const failed = await coordinator.waitForCompletion()
expect(failed).toMatchObject({
status: 'failed',
error: {
code: 'rate_limited',
retryable: true
}
})
expect(JSON.stringify(failed)).not.toContain('sk-private')
expect(failed?.progress).toEqual({
completed: 1,
total: 2,
percent: 50
})
expect(repository.records.get('document-1')?.[0]?.vector).toEqual([1, 0])
expect(repository.records.get('document-2')?.[0]?.vector).toEqual([0, 1])
expect(repository.errors.get('document-2')).toBe(
'向量服务当前请求过多。'
)
expect(repository.pendingRecords.size).toBe(0)
expect(JSON.stringify(repository.savedStatuses)).not.toContain(
'sk-private'
)
})
it('keeps completed documents and leaves unfinished documents unchanged when cancelled', async () => {
const repository = new MemoryRepository()
let receivedSignal: AbortSignal | undefined
const coordinator = new EmbeddingIndexCoordinator(repository, {
now: () => 30,
createId: () => 'job-cancelled'
})
await coordinator.initialize()
coordinator.startRebuild(
provider(
(input, signal) => {
if (input.includes('alpha')) {
return Promise.resolve(input.map(() => [1, 0]))
}
return new Promise<number[][]>((_resolve, reject) => {
receivedSignal = signal
signal?.addEventListener(
'abort',
() => reject(signal.reason),
{ once: true }
)
})
}
)
)
await vi.waitFor(() => {
expect(coordinator.status().job?.status).toBe('running')
expect(receivedSignal).toBeDefined()
})
expect(coordinator.cancel('another-job')).toBe(false)
expect(coordinator.cancel('job-cancelled')).toBe(true)
const cancelled = await coordinator.waitForCompletion()
expect(receivedSignal?.aborted).toBe(true)
expect(cancelled).toMatchObject({
status: 'cancelled',
completedAt: 30,
progress: { completed: 1, total: 2, percent: 50 }
})
expect(cancelled).not.toHaveProperty('error')
expect(repository.records.get('document-1')?.[0]?.vector).toEqual([1, 0])
expect(repository.records.get('document-2')?.[0]?.vector).toEqual([0, 1])
expect(repository.errors.has('document-2')).toBe(false)
expect(repository.pendingRecords.size).toBe(0)
expect(repository.lastJob).toEqual(cancelled)
})
it('marks an interrupted persisted job cancelled during initialization', async () => {
const repository = new MemoryRepository()
repository.lastJob = {
id: 'interrupted-job',
status: 'running',
provider: 'provider',
model: 'model',
progress: { completed: 1, total: 2, percent: 50 },
createdAt: 10,
startedAt: 11
}
const coordinator = new EmbeddingIndexCoordinator(repository, {
now: () => 12
})
await expect(coordinator.initialize()).resolves.toEqual({
job: expect.objectContaining({
id: 'interrupted-job',
status: 'cancelled',
completedAt: 12
})
})
expect(repository.lastJob).toMatchObject({
status: 'cancelled',
completedAt: 12
})
})
it('rejects overlapping rebuilds', async () => {
const repository = new MemoryRepository()
const coordinator = new EmbeddingIndexCoordinator(repository)
coordinator.startRebuild(
provider(
(_input, signal) =>
new Promise<number[][]>((_resolve, reject) => {
signal?.addEventListener('abort', () => reject(signal.reason), {
once: true
})
})
)
)
expect(() =>
coordinator.startRebuild(provider(async () => [[1, 0]]))
).toThrow('already active')
coordinator.cancel()
await coordinator.waitForCompletion()
})
})
@@ -0,0 +1,532 @@
import { randomUUID } from 'node:crypto'
import type {
EmbeddingDiagnosticResult,
EmbeddingIndexJob,
EmbeddingIndexStatus
} from '../../shared/embedding-contracts'
import {
classifyEmbeddingError,
EmbeddingOperationError
} from './embedding-errors'
const DEFAULT_BATCH_SIZE = 32
const MAX_BATCH_SIZE = 256
const MAX_VECTOR_DIMENSIONS = 8_192
export interface EmbeddingIndexProvider {
readonly provider: string
readonly model: string
readonly fingerprint?: string
embed(input: readonly string[], signal?: AbortSignal): Promise<number[][]>
}
export interface EmbeddingIndexItem {
id: string
content: string
contentChecksum?: string
}
export interface EmbeddingIndexRecord {
itemId: string
contentChecksum?: string
vector: readonly number[]
}
export interface EmbeddingIndexDocument {
id: string
items: readonly EmbeddingIndexItem[]
}
export interface EmbeddingIndexRepository {
getLastJob?(): Promise<EmbeddingIndexJob | null>
saveStatus?(status: EmbeddingIndexStatus): Promise<void>
listIndexDocumentIds(
signal: AbortSignal
): Promise<readonly string[]>
getIndexDocument(
documentId: string,
signal: AbortSignal
): Promise<EmbeddingIndexDocument | undefined>
beginDocumentReplacement(
documentId: string,
provider: string,
model: string,
signal: AbortSignal
): Promise<string>
appendDocumentReplacement(
replacementId: string,
documentId: string,
provider: string,
model: string,
records: readonly EmbeddingIndexRecord[],
signal: AbortSignal
): Promise<void>
finishDocumentReplacement(
replacementId: string,
documentId: string,
provider: string,
model: string,
signal: AbortSignal
): Promise<void>
discardDocumentReplacement(replacementId: string): Promise<void>
recordDocumentError(
documentId: string,
provider: string,
model: string,
error: string
): Promise<void>
}
export interface EmbeddingIndexCoordinatorOptions {
batchSize?: number
now?: () => number
createId?: () => string
}
export interface EmbeddingDiagnosticOptions {
signal?: AbortSignal
probeText?: string
}
export interface EmbeddingRebuildOptions {
signal?: AbortSignal
}
type JobListener = (status: EmbeddingIndexStatus) => void
function validatedLabel(value: string, name: string): string {
const normalized = value.trim()
if (normalized.length < 1 || normalized.length > 256) {
throw new RangeError(`${name} must contain between 1 and 256 characters`)
}
return normalized
}
function validateVector(
vector: readonly number[],
expectedDimensions?: number
): number {
if (
!Array.isArray(vector) ||
vector.length < 1 ||
vector.length > MAX_VECTOR_DIMENSIONS
) {
throw new EmbeddingOperationError({
code: 'invalid_response',
message: '向量服务返回了无效结果。',
retryable: false,
remedy: '请确认服务返回维度一致的有效向量。'
})
}
let magnitudeSquared = 0
for (const component of vector) {
if (typeof component !== 'number' || !Number.isFinite(component)) {
throw new EmbeddingOperationError({
code: 'invalid_response',
message: '向量服务返回了无效结果。',
retryable: false,
remedy: '请确认服务返回维度一致的有效向量。'
})
}
magnitudeSquared += component * component
}
if (
!Number.isFinite(magnitudeSquared) ||
magnitudeSquared <= 0 ||
(expectedDimensions !== undefined &&
vector.length !== expectedDimensions)
) {
throw new EmbeddingOperationError({
code: 'invalid_response',
message: '向量服务返回了无效结果。',
retryable: false,
remedy: '请确认服务返回维度一致的有效向量。'
})
}
return vector.length
}
function percent(completed: number, total: number): number {
return total === 0 ? 0 : (completed / total) * 100
}
export class EmbeddingIndexCoordinator {
private readonly repository: EmbeddingIndexRepository
private readonly batchSize: number
private readonly now: () => number
private readonly createId: () => string
private readonly listeners = new Set<JobListener>()
private job: EmbeddingIndexJob | null = null
private controller: AbortController | null = null
private completion: Promise<EmbeddingIndexJob> | null = null
private persistenceTail: Promise<void> = Promise.resolve()
constructor(
repository: EmbeddingIndexRepository,
options: EmbeddingIndexCoordinatorOptions = {}
) {
this.repository = repository
const batchSize = options.batchSize ?? DEFAULT_BATCH_SIZE
if (
!Number.isSafeInteger(batchSize) ||
batchSize < 1 ||
batchSize > MAX_BATCH_SIZE
) {
throw new RangeError(
`batchSize must be an integer between 1 and ${MAX_BATCH_SIZE}`
)
}
this.batchSize = batchSize
this.now = options.now ?? Date.now
this.createId = options.createId ?? randomUUID
}
async initialize(): Promise<EmbeddingIndexStatus> {
this.job = (await this.repository.getLastJob?.()) ?? null
if (
this.job?.status === 'queued' ||
this.job?.status === 'running'
) {
this.job = {
...this.job,
status: 'cancelled',
completedAt: this.now()
}
await this.persistStatus()
}
return this.status()
}
status(): EmbeddingIndexStatus {
return {
job: this.job
}
}
subscribe(listener: JobListener): () => void {
this.listeners.add(listener)
listener(this.status())
return () => {
this.listeners.delete(listener)
}
}
async diagnose(
provider: EmbeddingIndexProvider,
options: EmbeddingDiagnosticOptions = {}
): Promise<EmbeddingDiagnosticResult> {
const providerName = validatedLabel(provider.provider, 'provider')
const model = validatedLabel(provider.model, 'model')
const startedAt = this.now()
try {
const vectors = await provider.embed(
[options.probeText ?? 'GoodBuddy 向量模型连接测试'],
options.signal
)
if (vectors.length !== 1 || !vectors[0]) {
throw new EmbeddingOperationError({
code: 'invalid_response',
message: '向量服务返回了无效结果。',
retryable: false,
remedy: '请确认服务为每个输入返回一个有效向量。'
})
}
const dimensions = validateVector(vectors[0])
const checkedAt = this.now()
return {
status: 'available',
provider: providerName,
model,
checkedAt,
latencyMs: Math.max(0, checkedAt - startedAt),
dimensions
}
} catch (error) {
const checkedAt = this.now()
return {
status: 'unavailable',
provider: providerName,
model,
checkedAt,
latencyMs: Math.max(0, checkedAt - startedAt),
error:
error instanceof EmbeddingOperationError
? error.toSafeError()
: classifyEmbeddingError(error, {
cancelled: options.signal?.aborted
})
}
}
}
startRebuild(
provider: EmbeddingIndexProvider,
options: EmbeddingRebuildOptions = {}
): EmbeddingIndexJob {
if (
this.job?.status === 'queued' ||
this.job?.status === 'running'
) {
throw new Error('An embedding index rebuild is already active')
}
const providerName = validatedLabel(provider.provider, 'provider')
const model = validatedLabel(provider.model, 'model')
const controller = new AbortController()
const createdAt = this.now()
this.job = {
id: this.createId(),
status: 'queued',
provider: providerName,
model,
progress: { completed: 0, total: 0, percent: 0 },
createdAt
}
this.controller = controller
this.emit()
const externalSignal = options.signal
const forwardAbort = (): void => {
controller.abort(externalSignal?.reason)
}
if (externalSignal?.aborted) {
forwardAbort()
} else {
externalSignal?.addEventListener('abort', forwardAbort, {
once: true
})
}
this.completion = Promise.resolve()
.then(() => this.runRebuild(provider, controller.signal))
.finally(() => {
externalSignal?.removeEventListener('abort', forwardAbort)
if (this.controller === controller) {
this.controller = null
}
})
return this.job
}
async waitForCompletion(): Promise<EmbeddingIndexJob | null> {
return this.completion
}
cancel(jobId?: string): boolean {
if (
!this.controller ||
!this.job ||
(jobId !== undefined && this.job.id !== jobId) ||
!['queued', 'running'].includes(this.job.status)
) {
return false
}
this.controller.abort(new Error('Embedding index rebuild cancelled'))
return true
}
private async runRebuild(
provider: EmbeddingIndexProvider,
signal: AbortSignal
): Promise<EmbeddingIndexJob> {
try {
signal.throwIfAborted()
const documentIds =
await this.repository.listIndexDocumentIds(signal)
signal.throwIfAborted()
this.updateJob({
status: 'running',
startedAt: this.now(),
progress: {
completed: 0,
total: documentIds.length,
percent: 0
}
})
let completed = 0
let dimensions: number | undefined
for (const documentId of documentIds) {
signal.throwIfAborted()
const document = await this.repository.getIndexDocument(
documentId,
signal
)
if (!document) {
completed += 1
this.updateJob({
progress: {
completed,
total: documentIds.length,
percent: percent(completed, documentIds.length)
}
})
continue
}
const replacementId =
await this.repository.beginDocumentReplacement(
document.id,
provider.provider,
provider.model,
signal
)
try {
for (
let offset = 0;
offset < document.items.length;
offset += this.batchSize
) {
signal.throwIfAborted()
const batch = document.items.slice(
offset,
offset + this.batchSize
)
const vectors = await provider.embed(
batch.map((item) => item.content),
signal
)
if (vectors.length !== batch.length) {
throw new EmbeddingOperationError({
code: 'invalid_response',
message: '向量服务返回了无效结果。',
retryable: false,
remedy: '请确认服务为每个输入返回一个有效向量。'
})
}
const records = batch.map((item, index) => {
const vector = vectors[index]
if (!vector) {
throw new EmbeddingOperationError({
code: 'invalid_response',
message: '向量服务返回了无效结果。',
retryable: false
})
}
dimensions = validateVector(vector, dimensions)
return {
itemId: item.id,
...(item.contentChecksum
? { contentChecksum: item.contentChecksum }
: {}),
vector
}
})
await this.repository.appendDocumentReplacement(
replacementId,
document.id,
provider.provider,
provider.model,
records,
signal
)
}
signal.throwIfAborted()
await this.repository.finishDocumentReplacement(
replacementId,
document.id,
provider.provider,
provider.model,
signal
)
} catch (error) {
await this.repository
.discardDocumentReplacement(replacementId)
.catch(() => undefined)
if (signal.aborted) {
throw error
}
const safeError =
error instanceof EmbeddingOperationError
? error.toSafeError()
: classifyEmbeddingError(error)
await this.repository.recordDocumentError(
document.id,
provider.provider,
provider.model,
safeError.message
)
throw error
}
completed += 1
this.updateJob({
progress: {
completed,
total: documentIds.length,
percent: percent(completed, documentIds.length)
}
})
}
signal.throwIfAborted()
this.updateJob({
status: 'completed',
completedAt: this.now(),
progress: {
completed: documentIds.length,
total: documentIds.length,
percent: 100
}
})
} catch (error) {
const cancelled =
signal.aborted ||
classifyEmbeddingError(error).code === 'cancelled'
this.updateJob(
cancelled
? {
status: 'cancelled',
completedAt: this.now()
}
: {
status: 'failed',
completedAt: this.now(),
error:
error instanceof EmbeddingOperationError
? error.toSafeError()
: classifyEmbeddingError(error)
}
)
}
await this.persistenceTail
if (!this.job) {
throw new Error('Embedding index job state was lost')
}
return this.job
}
private updateJob(update: Partial<EmbeddingIndexJob>): void {
if (!this.job) {
throw new Error('No embedding index job is active')
}
this.job = {
...this.job,
...update
}
this.emit()
}
private emit(): void {
const status = this.status()
this.persistenceTail = this.persistenceTail.then(async () => {
try {
await this.repository.saveStatus?.(status)
} catch {
// Persistence failure must not interrupt an active provider operation.
}
})
for (const listener of this.listeners) {
listener(status)
}
}
private async persistStatus(): Promise<void> {
const status = this.status()
this.persistenceTail = this.persistenceTail.then(async () => {
try {
await this.repository.saveStatus?.(status)
} catch {
// Persistence failure must not interrupt initialization.
}
})
await this.persistenceTail
}
}
+213 -8
View File
@@ -86,12 +86,17 @@ describe('KnowledgeDatabase', () => {
const inspection = new DatabaseSync(path)
expect(
inspection.prepare('PRAGMA user_version').get()
).toEqual({ user_version: 2 })
).toEqual({ user_version: 4 })
expect(
inspection
.prepare('SELECT version FROM schema_migrations ORDER BY version')
.all()
).toEqual([{ version: 1 }, { version: 2 }])
).toEqual([
{ version: 1 },
{ version: 2 },
{ version: 3 },
{ version: 4 }
])
inspection.close()
const reopened = new KnowledgeDatabase(path)
@@ -108,7 +113,7 @@ describe('KnowledgeDatabase', () => {
.toHaveLength(1)
})
it('upgrades an existing v1 database to vector schema v2', async () => {
it('upgrades an existing v1 database to embedding rebuild schema v4', async () => {
const { database, path } = await createDatabase()
const knowledgeBase = database.createKnowledgeBase({
name: 'Version one data',
@@ -119,9 +124,11 @@ describe('KnowledgeDatabase', () => {
const downgrade = new DatabaseSync(path)
downgrade.exec(`
DROP TABLE embedding_rebuild_staging;
DROP TABLE embedding_index_job;
DROP TABLE embedding_index_state;
DROP TABLE chunk_embeddings;
DELETE FROM schema_migrations WHERE version = 2;
DELETE FROM schema_migrations WHERE version IN (2, 3, 4);
PRAGMA user_version = 1;
`)
downgrade.close()
@@ -131,20 +138,22 @@ describe('KnowledgeDatabase', () => {
upgraded.initialize()
const inspection = new DatabaseSync(path)
expect(inspection.prepare('PRAGMA user_version').get()).toEqual({
user_version: 2
user_version: 4
})
expect(
inspection
.prepare(
`SELECT name FROM sqlite_master
WHERE type = 'table' AND name IN
('chunk_embeddings', 'embedding_index_state')
WHERE type = 'table'
AND (name = 'chunk_embeddings' OR name LIKE 'embedding_%')
ORDER BY name`
)
.all()
).toEqual([
{ name: 'chunk_embeddings' },
{ name: 'embedding_index_state' }
{ name: 'embedding_index_job' },
{ name: 'embedding_index_state' },
{ name: 'embedding_rebuild_staging' }
])
inspection.close()
expect(
@@ -606,6 +615,202 @@ describe('KnowledgeDatabase', () => {
expect(database.graphSearch(first.id, unbacked.name)).toEqual([])
})
it('lists rebuild work by document and updates embedding state incrementally', async () => {
const { database } = await createDatabase()
const knowledgeBase = database.createKnowledgeBase({
name: 'Incremental index',
storageMode: 'reference',
graphEnabled: false
})
const alpha = seedDocument(database, knowledgeBase.id, 'incremental-alpha')
const beta = seedDocument(database, knowledgeBase.id, 'incremental-beta')
const documentIds = database.listEmbeddingIndexDocumentIds()
const documents = documentIds.map(
(documentId) =>
database.getEmbeddingIndexDocument(documentId)!
)
expect(documentIds).toEqual(
expect.arrayContaining([alpha.documentId, beta.documentId])
)
expect(documents).toEqual(
expect.arrayContaining([
expect.objectContaining({
id: alpha.documentId,
items: [
expect.objectContaining({
id: alpha.chunkId,
content: expect.stringContaining('lighthouse'),
contentChecksum: expect.stringMatching(/^[a-f0-9]{64}$/u)
})
]
}),
expect.objectContaining({
id: beta.documentId,
items: [
expect.objectContaining({
id: beta.chunkId,
content: expect.stringContaining('lighthouse'),
contentChecksum: expect.stringMatching(/^[a-f0-9]{64}$/u)
})
]
})
])
)
const alphaDocument = documents.find(
(document) => document.id === alpha.documentId
)!
const betaDocument = documents.find(
(document) => document.id === beta.documentId
)!
for (const [document, vector] of [
[alphaDocument, [1, 0]],
[betaDocument, [0, 1]]
] as const) {
database.replaceDocumentEmbeddings(
document.id,
'openai-compatible',
'embed-v1',
document.items.map((item) => ({
chunkId: item.id,
contentChecksum: item.contentChecksum!,
vector
}))
)
}
database.recordEmbeddingIndexError(
beta.documentId,
'openai-compatible',
'embed-v1',
'向量服务暂时不可用。'
)
expect(
database.getEmbeddingIndexState(
alpha.documentId,
'openai-compatible',
'embed-v1'
)
).toMatchObject({ status: 'ready' })
expect(
database.getEmbeddingIndexState(
beta.documentId,
'openai-compatible',
'embed-v1'
)
).toMatchObject({
status: 'error',
lastError: '向量服务暂时不可用。'
})
expect(
database
.vectorSearch({
knowledgeBaseId: knowledgeBase.id,
provider: 'openai-compatible',
model: 'embed-v1',
vector: [0, 1]
})
.map((result) => result.chunk.id)
).not.toContain(beta.chunkId)
})
it('stages embedding batches before atomically replacing a document index', async () => {
const { database } = await createDatabase()
const knowledgeBase = database.createKnowledgeBase({
name: 'Bounded rebuild',
storageMode: 'reference'
})
const seeded = seedDocument(
database,
knowledgeBase.id,
'bounded-rebuild'
)
const document =
database.getEmbeddingIndexDocument(seeded.documentId)!
const item = document.items[0]!
database.replaceDocumentEmbeddings(
document.id,
'openai-compatible',
'embed-v1',
[
{
chunkId: item.id,
contentChecksum: item.contentChecksum!,
vector: [1, 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: [0, 1]
}
]
)
expect(
database.vectorSearch({
knowledgeBaseId: knowledgeBase.id,
provider: 'openai-compatible',
model: 'embed-v1',
vector: [1, 0]
})[0]?.chunk.id
).toBe(item.id)
database.finishDocumentEmbeddingReplacement(
replacementId,
document.id,
'openai-compatible',
'embed-v1'
)
expect(
database.vectorSearch({
knowledgeBaseId: knowledgeBase.id,
provider: 'openai-compatible',
model: 'embed-v1',
vector: [0, 1]
})[0]?.chunk.id
).toBe(item.id)
})
it('persists the last embedding index job across restarts', async () => {
const created = await createDatabase()
let database = created.database
expect(database.getLastEmbeddingIndexJob()).toBeNull()
database.saveEmbeddingIndexJob({
id: 'job-1',
status: 'running',
provider: 'openai-compatible',
model: 'embed-v1',
progress: { completed: 1, total: 2, percent: 50 },
createdAt: 10,
startedAt: 11
})
database.close()
database = new KnowledgeDatabase(created.path)
openDatabases.push(database)
database.initialize()
expect(database.getLastEmbeddingIndexJob()).toMatchObject({
id: 'job-1',
status: 'running',
progress: { completed: 1, total: 2, percent: 50 }
})
database.saveEmbeddingIndexJob(null)
expect(database.getLastEmbeddingIndexJob()).toBeNull()
})
it('bounds inputs and rejects API keys in extensible metadata', async () => {
const { database } = await createDatabase()
expect(() =>
+457 -1
View File
@@ -1,5 +1,12 @@
import { createHash, randomUUID } from 'node:crypto'
import { DatabaseSync, type StatementSync } from 'node:sqlite'
import {
embeddingIndexJobSchema,
type EmbeddingIndexJob
} from '../../shared/embedding-contracts'
import type {
EmbeddingIndexDocument
} from './embedding-index-coordinator'
import type {
Chunk,
ChunkEmbeddingInput,
@@ -33,7 +40,7 @@ import type {
VectorSearchOptions
} from './types'
const DATABASE_VERSION = 2
const DATABASE_VERSION = 4
const MAX_ID_LENGTH = 128
const MAX_NAME_LENGTH = 512
const MAX_LOCATION_LENGTH = 8192
@@ -48,6 +55,7 @@ const MAX_JSON_DEPTH = 20
const MAX_JSON_NODES = 10_000
const MAX_JSON_STRING_LENGTH = 32_768
const MAX_EMBEDDING_DIMENSIONS = 8_192
const MAX_EMBEDDING_BATCH = 256
const MAX_EMBEDDING_PROVIDER_LENGTH = 128
const MAX_EMBEDDING_MODEL_LENGTH = 512
const MAX_EMBEDDING_ERROR_LENGTH = 2_000
@@ -478,6 +486,9 @@ export class KnowledgeDatabase {
`)
this.assertFts5(database)
this.migrate(database)
database
.prepare('DELETE FROM embedding_rebuild_staging')
.run()
this.database = database
} catch (error) {
database.close()
@@ -1052,6 +1063,310 @@ export class KnowledgeDatabase {
)
}
beginDocumentEmbeddingReplacement(
documentId: string,
provider: string,
model: string
): string {
const normalizedDocumentId = requiredString(
documentId,
'documentId',
MAX_ID_LENGTH
)
const normalizedProvider = requiredString(
provider,
'provider',
MAX_EMBEDDING_PROVIDER_LENGTH
)
const normalizedModel = requiredString(
model,
'model',
MAX_EMBEDDING_MODEL_LENGTH
)
const database = this.requireDatabase()
if (
!database
.prepare('SELECT 1 FROM documents WHERE id = ?')
.get(normalizedDocumentId)
) {
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()
}
appendDocumentEmbeddingBatch(
replacementId: string,
documentId: string,
provider: string,
model: string,
embeddings: readonly ChunkEmbeddingInput[]
): void {
const normalizedReplacementId = requiredString(
replacementId,
'replacementId',
MAX_ID_LENGTH
)
const normalizedDocumentId = requiredString(
documentId,
'documentId',
MAX_ID_LENGTH
)
const normalizedProvider = requiredString(
provider,
'provider',
MAX_EMBEDDING_PROVIDER_LENGTH
)
const normalizedModel = requiredString(
model,
'model',
MAX_EMBEDDING_MODEL_LENGTH
)
if (
!Array.isArray(embeddings) ||
embeddings.length < 1 ||
embeddings.length > MAX_EMBEDDING_BATCH
) {
throw new RangeError(
`embeddings must contain between 1 and ${MAX_EMBEDDING_BATCH} items`
)
}
const database = this.requireDatabase()
const findChunk = database.prepare(
'SELECT content FROM chunks WHERE id = ? AND document_id = ?'
)
const existingDimensions = database
.prepare(
`SELECT dimensions FROM embedding_rebuild_staging
WHERE replacement_id = ? LIMIT 1`
)
.get(normalizedReplacementId)
let dimensions = existingDimensions
? asNumber(existingDimensions, 'dimensions')
: undefined
const normalized = embeddings.map((embedding, index) => {
const chunkId = requiredString(
embedding.chunkId,
`embeddings[${index}].chunkId`,
MAX_ID_LENGTH
)
const chunk = findChunk.get(chunkId, normalizedDocumentId)
if (!chunk) {
throw new Error(
'Embeddings must reference chunks in the document'
)
}
const checksum = normalizedChecksum(
embedding.contentChecksum,
`embeddings[${index}].contentChecksum`
)
if (checksum !== contentChecksum(asString(chunk, 'content'))) {
throw new Error(
'Embedding content checksum does not match the chunk'
)
}
const vector = normalizeVector(
embedding.vector,
`embeddings[${index}].vector`
)
if (dimensions === undefined) {
dimensions = vector.dimensions
} else if (dimensions !== vector.dimensions) {
throw new Error(
'Document embeddings must have consistent dimensions'
)
}
return { chunkId, checksum, ...vector }
})
const insert = database.prepare(
`INSERT INTO embedding_rebuild_staging
(replacement_id, document_id, provider, model, chunk_id,
dimensions, content_checksum, vector, magnitude)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`
)
this.transaction(database, () => {
for (const item of normalized) {
insert.run(
normalizedReplacementId,
normalizedDocumentId,
normalizedProvider,
normalizedModel,
item.chunkId,
item.dimensions,
item.checksum,
item.bytes,
item.magnitude
)
}
})
}
finishDocumentEmbeddingReplacement(
replacementId: string,
documentId: string,
provider: string,
model: string
): EmbeddingIndexState {
const normalizedReplacementId = requiredString(
replacementId,
'replacementId',
MAX_ID_LENGTH
)
const normalizedDocumentId = requiredString(
documentId,
'documentId',
MAX_ID_LENGTH
)
const normalizedProvider = requiredString(
provider,
'provider',
MAX_EMBEDDING_PROVIDER_LENGTH
)
const normalizedModel = requiredString(
model,
'model',
MAX_EMBEDDING_MODEL_LENGTH
)
const database = this.requireDatabase()
const document = database
.prepare('SELECT knowledge_base_id FROM documents WHERE id = ?')
.get(normalizedDocumentId)
if (!document) {
throw new Error(`Document not found: ${normalizedDocumentId}`)
}
const counts = database
.prepare(
`SELECT
(SELECT COUNT(*) FROM chunks WHERE document_id = ?) AS chunks,
(SELECT COUNT(*) FROM embedding_rebuild_staging
WHERE replacement_id = ? AND document_id = ?
AND provider = ? AND model = ?) AS embeddings`
)
.get(
normalizedDocumentId,
normalizedReplacementId,
normalizedDocumentId,
normalizedProvider,
normalizedModel
)
if (
!counts ||
asNumber(counts, 'chunks') !== asNumber(counts, 'embeddings')
) {
throw new Error('Embeddings must cover every current document chunk')
}
const indexHash = createHash('sha256')
let dimensions: number | undefined
let firstChecksum = true
for (const row of database
.prepare(
`SELECT chunk_id, content_checksum, dimensions
FROM embedding_rebuild_staging
WHERE replacement_id = ? ORDER BY chunk_id`
)
.iterate(normalizedReplacementId)) {
const chunkId = asString(row, 'chunk_id')
const checksum = asString(row, 'content_checksum')
if (!firstChecksum) {
indexHash.update('\n')
}
indexHash.update(`${chunkId}\0${checksum}`)
firstChecksum = false
const rowDimensions = asNumber(row, 'dimensions')
if (dimensions === undefined) {
dimensions = rowDimensions
} else if (dimensions !== rowDimensions) {
throw new Error(
'Document embeddings must have consistent dimensions'
)
}
}
const now = new Date().toISOString()
this.transaction(database, () => {
database
.prepare(
`DELETE FROM chunk_embeddings
WHERE provider = ? AND model = ? AND chunk_id IN
(SELECT id FROM chunks WHERE document_id = ?)`
)
.run(normalizedProvider, normalizedModel, normalizedDocumentId)
database
.prepare(
`INSERT INTO chunk_embeddings
(chunk_id, knowledge_base_id, provider, model, dimensions,
content_checksum, vector, magnitude, created_at, updated_at)
SELECT chunk_id, ?, provider, model, dimensions,
content_checksum, vector, magnitude, ?, ?
FROM embedding_rebuild_staging
WHERE replacement_id = ?`
)
.run(
asString(document, 'knowledge_base_id'),
now,
now,
normalizedReplacementId
)
database
.prepare(
`INSERT INTO embedding_index_state
(document_id, knowledge_base_id, provider, model, dimensions,
content_checksum, status, last_error, updated_at)
VALUES (?, ?, ?, ?, ?, ?, 'ready', NULL, ?)
ON CONFLICT(document_id, provider, model) DO UPDATE SET
knowledge_base_id = excluded.knowledge_base_id,
dimensions = excluded.dimensions,
content_checksum = excluded.content_checksum,
status = 'ready',
last_error = NULL,
updated_at = excluded.updated_at`
)
.run(
normalizedDocumentId,
asString(document, 'knowledge_base_id'),
normalizedProvider,
normalizedModel,
dimensions ?? null,
indexHash.digest('hex'),
now
)
database
.prepare(
`DELETE FROM embedding_rebuild_staging
WHERE replacement_id = ?`
)
.run(normalizedReplacementId)
})
return this.requiredEmbeddingIndexState(
normalizedDocumentId,
normalizedProvider,
normalizedModel
)
}
discardDocumentEmbeddingReplacement(replacementId: string): void {
this.requireDatabase()
.prepare(
`DELETE FROM embedding_rebuild_staging
WHERE replacement_id = ?`
)
.run(
requiredString(
replacementId,
'replacementId',
MAX_ID_LENGTH
)
)
}
recordEmbeddingIndexError(
documentId: string,
provider: string,
@@ -1134,6 +1449,96 @@ export class KnowledgeDatabase {
return row ? mapEmbeddingIndexState(row) : undefined
}
getLastEmbeddingIndexJob(): EmbeddingIndexJob | null {
const row = this.requireDatabase()
.prepare(
'SELECT status_json FROM embedding_index_job WHERE singleton = 1'
)
.get()
if (!row) {
return null
}
try {
return embeddingIndexJobSchema.parse(
JSON.parse(asString(row, 'status_json'))
)
} catch {
return null
}
}
saveEmbeddingIndexJob(job: EmbeddingIndexJob | null): void {
const database = this.requireDatabase()
if (!job) {
database
.prepare('DELETE FROM embedding_index_job WHERE singleton = 1')
.run()
return
}
const normalized = embeddingIndexJobSchema.parse(job)
database
.prepare(
`INSERT INTO embedding_index_job
(singleton, status_json, updated_at)
VALUES (1, ?, ?)
ON CONFLICT(singleton) DO UPDATE SET
status_json = excluded.status_json,
updated_at = excluded.updated_at`
)
.run(JSON.stringify(normalized), new Date().toISOString())
}
listEmbeddingIndexDocumentIds(): string[] {
return this.requireDatabase()
.prepare(
`SELECT d.id
FROM documents d
WHERE json_extract(d.metadata, '$.status') IS NULL
OR json_extract(d.metadata, '$.status') = 'ready'
ORDER BY d.knowledge_base_id, d.id`
)
.all()
.map((document) => asString(document, 'id'))
}
getEmbeddingIndexDocument(
documentId: string
): EmbeddingIndexDocument | undefined {
const database = this.requireDatabase()
const normalizedDocumentId = requiredString(
documentId,
'documentId',
MAX_ID_LENGTH
)
const document = database
.prepare(
`SELECT d.id
FROM documents d
WHERE d.id = ?
AND (json_extract(d.metadata, '$.status') IS NULL
OR json_extract(d.metadata, '$.status') = 'ready')`
)
.get(normalizedDocumentId)
if (!document) {
return undefined
}
const chunks = database.prepare(
`SELECT id, content FROM chunks
WHERE document_id = ? ORDER BY ordinal ASC, id ASC`
)
return {
id: normalizedDocumentId,
items: chunks.all(normalizedDocumentId).map((row) => {
const content = asString(row, 'content')
return {
id: asString(row, 'id'),
content,
contentChecksum: contentChecksum(content)
}
})
}
}
vectorSearch(options: VectorSearchOptions): SearchResult[] {
return this.vectorSearchScored(options).map((item) => item.result)
}
@@ -2152,6 +2557,22 @@ export class KnowledgeDatabase {
)
.run(2, new Date().toISOString())
}
if (currentVersion < 3) {
this.migrateToVersion3(database)
database
.prepare(
'INSERT INTO schema_migrations (version, applied_at) VALUES (?, ?)'
)
.run(3, new Date().toISOString())
}
if (currentVersion < 4) {
this.migrateToVersion4(database)
database
.prepare(
'INSERT INTO schema_migrations (version, applied_at) VALUES (?, ?)'
)
.run(4, new Date().toISOString())
}
database.exec(`PRAGMA user_version = ${DATABASE_VERSION}`)
database.exec('COMMIT')
} catch (error) {
@@ -2332,6 +2753,41 @@ export class KnowledgeDatabase {
`)
}
private migrateToVersion3(database: DatabaseSync): void {
database.exec(`
CREATE TABLE embedding_index_job (
singleton INTEGER PRIMARY KEY CHECK (singleton = 1),
status_json TEXT NOT NULL CHECK (length(status_json) <= 32768),
updated_at TEXT NOT NULL
);
`)
}
private migrateToVersion4(database: DatabaseSync): void {
database.exec(`
CREATE TABLE embedding_rebuild_staging (
replacement_id TEXT NOT NULL,
document_id TEXT NOT NULL
REFERENCES documents(id) ON DELETE CASCADE,
provider TEXT NOT NULL,
model TEXT NOT NULL,
chunk_id TEXT NOT NULL
REFERENCES chunks(id) ON DELETE CASCADE,
dimensions INTEGER NOT NULL
CHECK (dimensions >= 1 AND dimensions <= 8192),
content_checksum TEXT NOT NULL
CHECK (length(content_checksum) = 64),
vector BLOB NOT NULL,
magnitude REAL NOT NULL CHECK (magnitude > 0),
PRIMARY KEY (replacement_id, chunk_id)
);
CREATE INDEX embedding_rebuild_staging_document_idx
ON embedding_rebuild_staging(
document_id, provider, model, replacement_id
);
`)
}
private normalizeChunks(chunks: ReplaceChunkInput[]): Array<{
id: string
ordinal: number
@@ -0,0 +1,117 @@
import type { EmbeddingIndexStatus } from '../../shared/embedding-contracts'
import type {
EmbeddingIndexRecord,
EmbeddingIndexRepository
} from './embedding-index-coordinator'
import type { KnowledgeDatabase } from './knowledge-database'
export class KnowledgeEmbeddingIndexRepository
implements EmbeddingIndexRepository {
constructor(private readonly database: KnowledgeDatabase) {}
async getLastJob(): Promise<EmbeddingIndexStatus['job']> {
return this.database.getLastEmbeddingIndexJob()
}
async saveStatus(status: EmbeddingIndexStatus): Promise<void> {
this.database.saveEmbeddingIndexJob(status.job)
}
async listIndexDocumentIds(signal: AbortSignal) {
signal.throwIfAborted()
const documentIds =
this.database.listEmbeddingIndexDocumentIds()
signal.throwIfAborted()
return documentIds
}
async getIndexDocument(
documentId: string,
signal: AbortSignal
) {
signal.throwIfAborted()
const document =
this.database.getEmbeddingIndexDocument(documentId)
signal.throwIfAborted()
return document
}
async beginDocumentReplacement(
documentId: string,
provider: string,
model: string,
signal: AbortSignal
): Promise<string> {
signal.throwIfAborted()
const replacementId =
this.database.beginDocumentEmbeddingReplacement(
documentId,
provider,
model
)
signal.throwIfAborted()
return replacementId
}
async appendDocumentReplacement(
replacementId: string,
documentId: string,
provider: string,
model: string,
records: readonly EmbeddingIndexRecord[],
signal: AbortSignal
): Promise<void> {
signal.throwIfAborted()
this.database.appendDocumentEmbeddingBatch(
replacementId,
documentId,
provider,
model,
records.map((record) => ({
chunkId: record.itemId,
contentChecksum: record.contentChecksum ?? '',
vector: record.vector
}))
)
signal.throwIfAborted()
}
async finishDocumentReplacement(
replacementId: string,
documentId: string,
provider: string,
model: string,
signal: AbortSignal
): Promise<void> {
signal.throwIfAborted()
this.database.finishDocumentEmbeddingReplacement(
replacementId,
documentId,
provider,
model
)
signal.throwIfAborted()
}
async discardDocumentReplacement(
replacementId: string
): Promise<void> {
this.database.discardDocumentEmbeddingReplacement(
replacementId
)
}
async recordDocumentError(
documentId: string,
provider: string,
model: string,
error: string
): Promise<void> {
this.database.recordEmbeddingIndexError(
documentId,
provider,
model,
error
)
}
}
+18 -4
View File
@@ -194,7 +194,10 @@ describe('KnowledgeService', () => {
provider: 'failing-provider',
model: 'failing-model',
embed: async () => {
throw new Error('synthetic provider outage')
throw Object.assign(
new Error('Bearer sk-private failed with private payload'),
{ status: 503 }
)
}
}
const { directory, service } = await createService(undefined, provider)
@@ -211,6 +214,8 @@ describe('KnowledgeService', () => {
if (!document) {
throw new Error('Indexed document missing')
}
expect(document.status).toBe('ready')
expect(service.snapshot(library.id).sources[0]?.status).toBe('ready')
expect(service.search(library.id, 'fallback')).toHaveLength(1)
expect(
service.database.getEmbeddingIndexState(
@@ -220,13 +225,22 @@ describe('KnowledgeService', () => {
)
).toMatchObject({
status: 'error',
lastError: 'synthetic provider outage'
lastError: '向量服务暂时不可用。'
})
expect(
JSON.stringify(
service.database.getEmbeddingIndexState(
document.id,
provider.provider,
provider.model
)
)
).not.toContain('sk-private')
const results = await service.searchHybrid(library.id, 'fallback')
expect(results[0]?.retrieval.channels).toContain('fts')
})
it('reindexes existing documents when an embedding provider is enabled', async () => {
it('defers existing-document rebuilds when an embedding provider is enabled', async () => {
const { directory, service } = await createService()
const sourcePath = join(directory, 'existing.txt')
await writeFile(sourcePath, 'existing semantic content', 'utf8')
@@ -251,7 +265,7 @@ describe('KnowledgeService', () => {
provider.provider,
provider.model
)
).toMatchObject({ status: 'ready', dimensions: 2 })
).toBeUndefined()
})
it('embeds a hybrid query once across multiple libraries', async () => {
+4 -50
View File
@@ -19,6 +19,7 @@ import {
resolve
} from 'node:path'
import { chunkDocument, parseDocument, supportedDocumentExtensions } from './document-parser'
import { classifyEmbeddingError } from './embedding-errors'
import {
extractKnowledgeGraph,
normalizeEntityAlias,
@@ -178,54 +179,8 @@ export class KnowledgeService {
}
setEmbeddingProvider(provider?: EmbeddingProvider): Promise<void> {
if (
this.embeddingProvider === provider ||
(this.embeddingProvider?.fingerprint !== undefined &&
this.embeddingProvider.fingerprint === provider?.fingerprint)
) {
this.embeddingProvider = provider
return Promise.resolve()
}
this.embeddingProvider = provider
if (!provider) {
return Promise.resolve()
}
const reindex = this.reindexEmbeddings(provider)
this.activeSyncs.set('embedding-reindex', reindex)
void reindex.then(
() => {
if (this.activeSyncs.get('embedding-reindex') === reindex) {
this.activeSyncs.delete('embedding-reindex')
}
},
() => {
if (this.activeSyncs.get('embedding-reindex') === reindex) {
this.activeSyncs.delete('embedding-reindex')
}
}
)
return reindex
}
private async reindexEmbeddings(
provider: EmbeddingProvider
): Promise<void> {
for (const library of this.database.listKnowledgeBases(100)) {
if (this.embeddingProvider !== provider) {
return
}
for (const document of this.database.listDocuments(
library.id,
500
)) {
if (this.embeddingProvider !== provider) {
return
}
if (document.metadata.status === 'ready') {
await this.indexDocumentEmbeddings(document, provider)
}
}
}
return Promise.resolve()
}
createLibrary(input: CreateKnowledgeBaseInput): KnowledgeBase {
@@ -780,14 +735,13 @@ export class KnowledgeService {
if (this.lifecycleController.signal.aborted) {
return
}
const message =
error instanceof Error ? error.message : 'Embedding indexing failed'
const safeError = classifyEmbeddingError(error)
try {
this.database.recordEmbeddingIndexError(
document.id,
provider.provider,
provider.model,
message.slice(0, 2_000)
safeError.message
)
} catch {
// FTS indexing is authoritative; embedding diagnostics are best effort.
+72 -1
View File
@@ -1,4 +1,5 @@
import { describe, expect, it, vi } from 'vitest'
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
import { setIntranetCompatibilityReader } from '../intranet-compatibility-policy'
import {
isPublicAddress,
normalizeSourceUrl,
@@ -7,6 +8,14 @@ import {
const publicAddress = [{ address: '93.184.216.34', family: 4 }]
beforeEach(() => {
setIntranetCompatibilityReader(() => false)
})
afterEach(() => {
setIntranetCompatibilityReader(() => true)
})
describe('URL importer', () => {
it('rejects local protocols, hosts and private address ranges', async () => {
expect(() => normalizeSourceUrl('file:///etc/passwd')).toThrow('HTTP')
@@ -16,6 +25,10 @@ describe('URL importer', () => {
expect(isPublicAddress('127.0.0.1')).toBe(false)
expect(isPublicAddress('10.0.0.1')).toBe(false)
expect(isPublicAddress('169.254.169.254')).toBe(false)
expect(isPublicAddress('192.0.2.1')).toBe(false)
expect(isPublicAddress('198.18.0.1')).toBe(false)
expect(isPublicAddress('198.51.100.1')).toBe(false)
expect(isPublicAddress('203.0.113.1')).toBe(false)
expect(isPublicAddress('::1')).toBe(false)
expect(isPublicAddress('fc00::1')).toBe(false)
expect(isPublicAddress('93.184.216.34')).toBe(true)
@@ -42,6 +55,64 @@ describe('URL importer', () => {
).rejects.toThrow('私网')
})
it('imports private intranet URLs in compatibility mode', async () => {
setIntranetCompatibilityReader(() => true)
const transport = vi.fn(async () => ({
status: 200,
headers: { 'content-type': 'text/plain' },
body: Buffer.from('内部知识')
}))
const importer = new UrlImporter({
lookup: async () => [{ address: '192.168.10.25', family: 4 }],
transport
})
await expect(
importer.import(
'http://knowledge.internal/guide',
new AbortController().signal
)
).resolves.toMatchObject({
url: 'http://knowledge.internal/guide',
contentType: 'text/plain'
})
expect(transport).toHaveBeenCalledWith(
expect.objectContaining({ hostname: 'knowledge.internal' }),
{ address: '192.168.10.25', family: 4 },
expect.any(AbortSignal),
expect.any(Number)
)
})
it('keeps metadata, link-local and mixed answers blocked in compatibility mode', async () => {
setIntranetCompatibilityReader(() => true)
expect(() =>
normalizeSourceUrl('http://metadata.google.internal/latest')
).toThrow('不允许')
expect(() =>
normalizeSourceUrl('http://user:secret@knowledge.internal')
).toThrow('不允许')
for (const addresses of [
[{ address: '169.254.169.254', family: 4 }],
[
{ address: '10.0.0.2', family: 4 },
{ address: '93.184.216.34', family: 4 }
]
]) {
const importer = new UrlImporter({
lookup: async () => addresses,
transport: vi.fn()
})
await expect(
importer.import(
'http://knowledge.internal',
new AbortController().signal
)
).rejects.toThrow('私网')
}
})
it('imports HTML and discovers only same-origin links', async () => {
const transport = vi.fn(async () => ({
status: 200,
+47 -50
View File
@@ -2,6 +2,11 @@ import { lookup as dnsLookup } from 'node:dns/promises'
import { request as httpRequest } from 'node:http'
import { isIP } from 'node:net'
import { request as httpsRequest } from 'node:https'
import { isIntranetCompatibilityEnabled } from '../intranet-compatibility-policy'
import {
isIntranetBrowserAddress,
isPublicBrowserAddress
} from '../browser/browser-url-policy'
import { parseDocument, type ParsedDocument } from './document-parser'
type ResolvedAddress = {
@@ -38,52 +43,28 @@ export type UrlImporterOptions = {
}
const blockedHostnames = new Set([
'localhost',
'localhost.localdomain',
'instance-data',
'instance-data.ec2.internal',
'metadata',
'metadata.aws.internal',
'metadata.google.internal'
])
function isPrivateIpv4(address: string): boolean {
const parts = address.split('.').map(Number)
if (parts.length !== 4 || parts.some((part) => !Number.isInteger(part))) {
return true
}
const [first = 0, second = 0] = parts
return (
first === 0 ||
first === 10 ||
first === 127 ||
(first === 169 && second === 254) ||
(first === 172 && second >= 16 && second <= 31) ||
(first === 192 && second === 168) ||
(first === 100 && second >= 64 && second <= 127) ||
first >= 224
)
}
function isPrivateIpv6(address: string): boolean {
const normalized = address.toLowerCase().split('%')[0] ?? ''
if (
normalized === '::' ||
normalized === '::1' ||
normalized.startsWith('fc') ||
normalized.startsWith('fd') ||
/^fe[89ab]/.test(normalized) ||
normalized.startsWith('ff')
) {
return true
}
const mapped = normalized.match(/^::ffff:(\d+\.\d+\.\d+\.\d+)$/)
return mapped ? isPrivateIpv4(mapped[1] ?? '') : false
}
export function isPublicAddress(address: string): boolean {
const family = isIP(address)
return family === 4
? !isPrivateIpv4(address)
: family === 6
? !isPrivateIpv6(address)
: false
return isPublicBrowserAddress(address)
}
export function isIntranetAddress(address: string): boolean {
return isIntranetBrowserAddress(address)
}
function addressClass(
address: string
): 'public' | 'intranet' | 'blocked' {
if (isPublicAddress(address)) {
return 'public'
}
return isIntranetAddress(address) ? 'intranet' : 'blocked'
}
export function normalizeSourceUrl(input: string): URL {
@@ -96,11 +77,19 @@ export function normalizeSourceUrl(input: string): URL {
if (!['http:', 'https:'].includes(url.protocol)) {
throw new Error('网页来源仅支持 HTTP(S)')
}
const hostname = url.hostname.toLowerCase().replace(/\.$/u, '')
if (
url.username ||
url.password ||
blockedHostnames.has(url.hostname.toLowerCase()) ||
url.hostname.toLowerCase().endsWith('.localhost')
blockedHostnames.has(hostname) ||
(
!isIntranetCompatibilityEnabled() &&
(
hostname === 'localhost' ||
hostname === 'localhost.localdomain' ||
hostname.endsWith('.localhost')
)
)
) {
throw new Error('该网页地址不允许导入')
}
@@ -211,15 +200,23 @@ export class UrlImporter {
this.maximumRedirects = options.maximumRedirects ?? 5
}
private async resolvePublic(url: URL): Promise<ResolvedAddress> {
private async resolveAddress(url: URL): Promise<ResolvedAddress> {
const addresses = await this.lookup(url.hostname)
const address = addresses.find((candidate) =>
isPublicAddress(candidate.address)
const classes = addresses.map((candidate) =>
candidate.family === isIP(candidate.address)
? addressClass(candidate.address)
: 'blocked'
)
const address = addresses[0]
if (
addresses.length === 0 ||
addresses.some((candidate) => !isPublicAddress(candidate.address)) ||
!address
!address ||
classes.includes('blocked') ||
new Set(classes).size !== 1 ||
(
!isIntranetCompatibilityEnabled() &&
classes.some((addressType) => addressType !== 'public')
)
) {
throw new Error('网页地址解析到本机、私网或不可用地址')
}
@@ -232,7 +229,7 @@ export class UrlImporter {
for (let redirect = 0; redirect <= this.maximumRedirects; redirect += 1) {
signal.throwIfAborted()
const address = await this.resolvePublic(url)
const address = await this.resolveAddress(url)
response = await this.transport(
url,
address,