mirror of
https://github.com/hellodigua/ChatLab.git
synced 2026-10-26 02:04:37 +08:00
fix(semantic-index): honor proxy for local model downloads
Pass a model download proxy through the semantic index worker and local embedding provider so local Hugging Face downloads use the effective proxy. Restart idle local workers when the proxy changes before preload, build, or search calls, and initialize app logging inside the worker process for preload diagnostics.
This commit is contained in:
1 parent
e5954d8fb1
commit
f304146c8f
11 files changed
+328
-17
No files matched your search
@@ -20,6 +20,7 @@
|
||||
"sqlite-vec": "^0.1.9",
|
||||
"stream-chain": "^2.2.5",
|
||||
"stream-json": "^1.9.1",
|
||||
"undici": "^6.25.0",
|
||||
"yauzl": "^3.4.0"
|
||||
},
|
||||
"optionalDependencies": {
|
||||
|
||||
@@ -16,13 +16,22 @@ const localConfig: SemanticIndexConfig = {
|
||||
}
|
||||
|
||||
test('builds local provider from profile with injected pipeline factory', async () => {
|
||||
let seenProxyUrl: string | undefined
|
||||
const fakeFactory: LocalPipelineFactory = async () => async (texts) =>
|
||||
texts.map(() => new Array(QWEN3_PROFILE.dim).fill(0.1))
|
||||
const embedder = createEmbedder(localConfig, { localPipelineFactory: fakeFactory })
|
||||
const fakeFactoryWithCapture: LocalPipelineFactory = async (params) => {
|
||||
seenProxyUrl = params.modelDownloadProxyUrl
|
||||
return fakeFactory(params)
|
||||
}
|
||||
const embedder = createEmbedder(localConfig, {
|
||||
localPipelineFactory: fakeFactoryWithCapture,
|
||||
modelDownloadProxyUrl: 'http://127.0.0.1:7890',
|
||||
})
|
||||
assert.equal(embedder.modelId, QWEN3_PROFILE.modelId)
|
||||
assert.equal(embedder.dim, QWEN3_PROFILE.dim)
|
||||
const [vector] = await embedder.embedDocuments(['hello'])
|
||||
assert.equal(vector.length, QWEN3_PROFILE.dim)
|
||||
assert.equal(seenProxyUrl, 'http://127.0.0.1:7890')
|
||||
})
|
||||
|
||||
test('throws for unknown local model', () => {
|
||||
|
||||
@@ -18,6 +18,8 @@ import type { SemanticIndexConfig } from './config'
|
||||
export interface EmbedderFactoryDeps {
|
||||
/** 本地模型目录,例如 ~/.chatlab/ai/models/semantic-index */
|
||||
modelsCacheDir?: string
|
||||
/** Optional HTTP(S) proxy URL used only for downloading local embedding model files. */
|
||||
modelDownloadProxyUrl?: string
|
||||
/** auth-profiles 解析,默认走 @openchatlab/config */
|
||||
resolveApiKey?: (provider: string, authProfile?: string) => string
|
||||
/** 本地 pipeline 工厂注入(测试用,不下载模型) */
|
||||
@@ -45,6 +47,7 @@ export function createEmbedder(config: SemanticIndexConfig, deps: EmbedderFactor
|
||||
}
|
||||
return new LocalEmbeddingProvider(profile, {
|
||||
cacheDir: deps.modelsCacheDir,
|
||||
modelDownloadProxyUrl: deps.modelDownloadProxyUrl,
|
||||
pipelineFactory: deps.localPipelineFactory,
|
||||
})
|
||||
}
|
||||
@@ -12,6 +12,10 @@ import type { EmbeddingProvider } from './types'
|
||||
import type { LocalEmbeddingProfile } from './profiles'
|
||||
import { applyQueryInstruction, clampTextChars } from './text'
|
||||
|
||||
type UndiciFetch = (typeof import('undici'))['fetch']
|
||||
type UndiciRequestInit = NonNullable<Parameters<UndiciFetch>[1]>
|
||||
const SOCKS_PROXY_PROTOCOLS = new Set(['socks:', 'socks4:', 'socks5:'])
|
||||
|
||||
/** 一次特征抽取调用:输入文本数组,返回每条文本的向量(number[][]) */
|
||||
export type FeatureExtractFn = (
|
||||
texts: string[],
|
||||
@@ -23,14 +27,33 @@ export type LocalPipelineFactory = (params: {
|
||||
modelId: string
|
||||
dtype?: 'fp32' | 'q8'
|
||||
cacheDir?: string
|
||||
modelDownloadProxyUrl?: string
|
||||
}) => Promise<FeatureExtractFn>
|
||||
|
||||
const defaultPipelineFactory: LocalPipelineFactory = async ({ modelId, dtype, cacheDir }) => {
|
||||
export async function createProxyFetch(proxyUrl: string): Promise<typeof fetch> {
|
||||
const parsed = new URL(proxyUrl)
|
||||
if (SOCKS_PROXY_PROTOCOLS.has(parsed.protocol)) {
|
||||
throw new Error(
|
||||
`SOCKS proxy is not supported for local embedding model downloads: ${parsed.protocol}//${parsed.host}`
|
||||
)
|
||||
}
|
||||
const { fetch: undiciFetch, ProxyAgent } = await import('undici')
|
||||
const dispatcher = new ProxyAgent(proxyUrl)
|
||||
return ((input, init) => {
|
||||
const nextInit: UndiciRequestInit = { ...(init as UndiciRequestInit | undefined), dispatcher }
|
||||
return undiciFetch(input as Parameters<UndiciFetch>[0], nextInit) as unknown as ReturnType<typeof fetch>
|
||||
}) as typeof fetch
|
||||
}
|
||||
|
||||
const defaultPipelineFactory: LocalPipelineFactory = async ({ modelId, dtype, cacheDir, modelDownloadProxyUrl }) => {
|
||||
const transformers = await import('@huggingface/transformers')
|
||||
if (cacheDir) {
|
||||
transformers.env.cacheDir = cacheDir
|
||||
transformers.env.allowRemoteModels = true
|
||||
}
|
||||
if (modelDownloadProxyUrl) {
|
||||
transformers.env.fetch = await createProxyFetch(modelDownloadProxyUrl)
|
||||
}
|
||||
const extractor = await transformers.pipeline('feature-extraction', modelId, dtype ? { dtype } : undefined)
|
||||
return async (texts, options) => {
|
||||
const out = await extractor(texts, { pooling: options.pooling, normalize: options.normalize })
|
||||
@@ -40,6 +63,7 @@ const defaultPipelineFactory: LocalPipelineFactory = async ({ modelId, dtype, ca
|
||||
|
||||
export interface LocalEmbeddingProviderOptions {
|
||||
cacheDir?: string
|
||||
modelDownloadProxyUrl?: string
|
||||
pipelineFactory?: LocalPipelineFactory
|
||||
}
|
||||
|
||||
@@ -69,6 +93,7 @@ export class LocalEmbeddingProvider implements EmbeddingProvider {
|
||||
modelId: this.profile.modelId,
|
||||
dtype: this.profile.dtype,
|
||||
cacheDir: this.options.cacheDir,
|
||||
modelDownloadProxyUrl: this.options.modelDownloadProxyUrl,
|
||||
})
|
||||
}
|
||||
return this.extractorPromise
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import assert from 'node:assert/strict'
|
||||
import test from 'node:test'
|
||||
import { LocalEmbeddingProvider, type LocalPipelineFactory } from './local'
|
||||
import { createProxyFetch, LocalEmbeddingProvider, type LocalPipelineFactory } from './local'
|
||||
import { OpenAICompatibleEmbeddingProvider, type FetchFn } from './api'
|
||||
import { QWEN3_PROFILE, type LocalEmbeddingProfile } from './profiles'
|
||||
|
||||
@@ -61,6 +61,10 @@ test('local provider throws when returned dim mismatches profile', async () => {
|
||||
await assert.rejects(() => provider.embedDocuments(['a']), /dim/i)
|
||||
})
|
||||
|
||||
test('local model download proxy rejects SOCKS URLs explicitly', async () => {
|
||||
await assert.rejects(() => createProxyFetch('socks5://127.0.0.1:1080'), /SOCKS proxy is not supported/)
|
||||
})
|
||||
|
||||
test('API provider posts OpenAI-compatible request and parses ordered embeddings', async () => {
|
||||
let captured: { url: string; body: string; headers: Record<string, string> } | null = null
|
||||
const fetchFn: FetchFn = async (url, init) => {
|
||||
|
||||
@@ -53,6 +53,7 @@ import type { SessionRuntimeAdapter } from '../services/adapters'
|
||||
import { clampSearchMaxResults, SEARCH_MAX_RESULTS_HARD_CAP } from './config'
|
||||
import { applyPreprocessingPipeline } from '../ai/preprocessor/preprocessing-pipeline'
|
||||
import type { PreprocessConfig, PreprocessableMessage } from '../ai/preprocessor/types'
|
||||
import { appLogger } from '../logging/app-logger'
|
||||
|
||||
export interface SemanticIndexServiceOptions {
|
||||
/** embedding_index.db 路径({vectorDir}/embedding_index.db) */
|
||||
@@ -62,6 +63,8 @@ export interface SemanticIndexServiceOptions {
|
||||
sessionAdapter: SessionRuntimeAdapter
|
||||
/** 本地模型缓存目录 */
|
||||
modelsCacheDir?: string
|
||||
/** Optional HTTP(S) proxy URL used only for local embedding model downloads. */
|
||||
modelDownloadProxyUrl?: string
|
||||
/** embedder 工厂注入(测试/平台) */
|
||||
embedderFactoryDeps?: EmbedderFactoryDeps
|
||||
/** sqlite-vec 加载器注入(Electron 打包) */
|
||||
@@ -206,12 +209,14 @@ export function createSemanticIndexService(params: {
|
||||
nativeBinding?: string
|
||||
loadSqliteVec?: LoadSqliteVec
|
||||
embedderFactoryDeps?: EmbedderFactoryDeps
|
||||
modelDownloadProxyUrl?: string
|
||||
}): SemanticIndexService {
|
||||
const { pathProvider } = params
|
||||
return new SemanticIndexService({
|
||||
vectorDbPath: path.join(pathProvider.getVectorDir(), SEMANTIC_INDEX_DB_FILE),
|
||||
configPath: path.join(pathProvider.getAiDataDir(), SEMANTIC_INDEX_CONFIG_FILE),
|
||||
modelsCacheDir: path.join(pathProvider.getAiDataDir(), 'models', 'semantic-index'),
|
||||
modelDownloadProxyUrl: params.modelDownloadProxyUrl,
|
||||
sessionAdapter: params.sessionAdapter,
|
||||
nativeBinding: params.nativeBinding,
|
||||
loadSqliteVec: params.loadSqliteVec,
|
||||
@@ -297,14 +302,19 @@ export class SemanticIndexService {
|
||||
}
|
||||
// 本地模式已配置且功能已开启时,立即在后台触发模型下载,让用户在建索引前完成等待
|
||||
if (saved.enabled && isSemanticIndexConfigured(saved) && saved.mode === 'local') {
|
||||
const modelId = saved.local.modelId
|
||||
const cacheDir = this.options.modelsCacheDir
|
||||
this.modelPreloadStatus = 'downloading'
|
||||
appLogger.info('semantic-index', 'local embedding model preload started', { modelId, cacheDir })
|
||||
void this.getEmbedder()
|
||||
.preload?.()
|
||||
.then(() => {
|
||||
this.modelPreloadStatus = 'ready'
|
||||
appLogger.info('semantic-index', 'local embedding model preload completed', { modelId, cacheDir })
|
||||
})
|
||||
.catch(() => {
|
||||
.catch((error) => {
|
||||
this.modelPreloadStatus = 'error'
|
||||
appLogger.error('semantic-index', `local embedding model preload failed: ${modelId}`, error)
|
||||
})
|
||||
}
|
||||
return saved
|
||||
@@ -319,6 +329,7 @@ export class SemanticIndexService {
|
||||
if (!this.embedder || this.embedderModelId !== modelId) {
|
||||
this.embedder = createEmbedder(this.configStore.get(), {
|
||||
modelsCacheDir: this.options.modelsCacheDir,
|
||||
modelDownloadProxyUrl: this.options.modelDownloadProxyUrl,
|
||||
...this.options.embedderFactoryDeps,
|
||||
})
|
||||
this.embedderModelId = modelId
|
||||
|
||||
@@ -43,9 +43,11 @@ class FakeTransport implements SemanticIndexWorkerTransport {
|
||||
|
||||
function makeFactory(
|
||||
instances: FakeTransport[],
|
||||
handler?: (method: string, args: unknown[]) => unknown
|
||||
handler?: (method: string, args: unknown[]) => unknown,
|
||||
proxyUrls?: Array<string | undefined>
|
||||
): SemanticIndexWorkerTransportFactory {
|
||||
return () => {
|
||||
return (modelDownloadProxyUrl) => {
|
||||
proxyUrls?.push(modelDownloadProxyUrl)
|
||||
const transport = new FakeTransport(handler)
|
||||
instances.push(transport)
|
||||
return transport
|
||||
@@ -190,6 +192,181 @@ test('worker client forwards config updates to a live worker without starting a
|
||||
])
|
||||
})
|
||||
|
||||
test('worker client restarts live worker before local model preload when proxy changes', async () => {
|
||||
const instances: FakeTransport[] = []
|
||||
const proxy = { url: undefined as string | undefined }
|
||||
const client = createSemanticIndexWorkerClient({
|
||||
configStore: makeConfigStore(),
|
||||
transportFactory: makeFactory(instances),
|
||||
getModelDownloadProxyUrl: () => proxy.url,
|
||||
idleTimeoutMs: 1000,
|
||||
})
|
||||
|
||||
await client.status('session-a')
|
||||
proxy.url = 'http://127.0.0.1:7890'
|
||||
const initial = await client.getConfig()
|
||||
const updated: SemanticIndexConfig = {
|
||||
...initial,
|
||||
enabled: true,
|
||||
mode: 'local',
|
||||
local: { modelId: 'model-a' },
|
||||
api: null,
|
||||
}
|
||||
|
||||
const saved = await client.setConfig(updated)
|
||||
|
||||
assert.equal(saved.local.modelId, 'model-a')
|
||||
assert.equal(instances.length, 2)
|
||||
assert.equal(instances[0].closed, true)
|
||||
assert.deepEqual(instances[0].requests, [{ method: 'status', args: ['session-a'] }])
|
||||
assert.deepEqual(instances[1].requests, [{ method: 'setConfig', args: [saved] }])
|
||||
})
|
||||
|
||||
test('worker client resolves async model download proxy before starting worker', async () => {
|
||||
const instances: FakeTransport[] = []
|
||||
const proxyUrls: Array<string | undefined> = []
|
||||
const client = createSemanticIndexWorkerClient({
|
||||
configStore: makeConfigStore(),
|
||||
transportFactory: makeFactory(instances, undefined, proxyUrls),
|
||||
getModelDownloadProxyUrl: async () => 'http://127.0.0.1:7890',
|
||||
idleTimeoutMs: 1000,
|
||||
})
|
||||
|
||||
await client.status('session-a')
|
||||
|
||||
assert.equal(instances.length, 1)
|
||||
assert.deepEqual(proxyUrls, ['http://127.0.0.1:7890'])
|
||||
})
|
||||
|
||||
test('worker client restarts live worker before local build calls when proxy changes', async () => {
|
||||
const cases: Array<{
|
||||
action: 'build' | 'rebuild' | 'buildAllPending'
|
||||
run: (client: ReturnType<typeof createSemanticIndexWorkerClient>) => Promise<void>
|
||||
expectedRequest: { method: string; args: unknown[] }
|
||||
}> = [
|
||||
{
|
||||
action: 'build',
|
||||
run: (client) => client.build('session-a'),
|
||||
expectedRequest: { method: 'build', args: ['session-a'] },
|
||||
},
|
||||
{
|
||||
action: 'rebuild',
|
||||
run: (client) => client.rebuild('session-a'),
|
||||
expectedRequest: { method: 'rebuild', args: ['session-a'] },
|
||||
},
|
||||
{
|
||||
action: 'buildAllPending',
|
||||
run: (client) => client.buildAllPending(),
|
||||
expectedRequest: { method: 'buildAllPending', args: [] },
|
||||
},
|
||||
]
|
||||
|
||||
for (const { action, run, expectedRequest } of cases) {
|
||||
const instances: FakeTransport[] = []
|
||||
const proxy = { url: undefined as string | undefined }
|
||||
const configStore = makeConfigStore()
|
||||
configStore.set({ version: 1, enabled: true, mode: 'local', local: { modelId: 'model-a' }, api: null })
|
||||
const client = createSemanticIndexWorkerClient({
|
||||
configStore,
|
||||
transportFactory: makeFactory(instances, (method) => {
|
||||
if (method === 'status') return null
|
||||
if (method === action) return undefined
|
||||
throw new Error(`unexpected method: ${method}`)
|
||||
}),
|
||||
getModelDownloadProxyUrl: () => proxy.url,
|
||||
idleTimeoutMs: 1000,
|
||||
})
|
||||
|
||||
await client.status('session-a')
|
||||
proxy.url = 'http://127.0.0.1:7890'
|
||||
await run(client)
|
||||
|
||||
assert.equal(instances.length, 2, action)
|
||||
assert.equal(instances[0].closed, true, action)
|
||||
assert.deepEqual(instances[0].requests, [{ method: 'status', args: ['session-a'] }], action)
|
||||
assert.deepEqual(instances[1].requests, [expectedRequest], action)
|
||||
}
|
||||
})
|
||||
|
||||
test('worker client restarts live worker before local search calls when proxy changes', async () => {
|
||||
const cases: Array<{
|
||||
action: 'search' | 'searchForTool'
|
||||
run: (client: ReturnType<typeof createSemanticIndexWorkerClient>) => Promise<unknown>
|
||||
expectedRequest: { method: string; args: unknown[] }
|
||||
}> = [
|
||||
{
|
||||
action: 'search',
|
||||
run: (client) => client.search('session-a', 'query-a', { finalTopK: 3 }),
|
||||
expectedRequest: { method: 'search', args: ['session-a', 'query-a', { finalTopK: 3 }] },
|
||||
},
|
||||
{
|
||||
action: 'searchForTool',
|
||||
run: (client) => client.searchForTool('session-a', 'query-a', { maxResults: 3 }),
|
||||
expectedRequest: { method: 'searchForTool', args: ['session-a', 'query-a', { maxResults: 3 }] },
|
||||
},
|
||||
]
|
||||
|
||||
for (const { action, run, expectedRequest } of cases) {
|
||||
const instances: FakeTransport[] = []
|
||||
const proxy = { url: undefined as string | undefined }
|
||||
const configStore = makeConfigStore()
|
||||
configStore.set({ version: 1, enabled: true, mode: 'local', local: { modelId: 'model-a' }, api: null })
|
||||
const client = createSemanticIndexWorkerClient({
|
||||
configStore,
|
||||
transportFactory: makeFactory(instances, (method) => {
|
||||
if (method === 'status') return null
|
||||
if (method === action) return {}
|
||||
throw new Error(`unexpected method: ${method}`)
|
||||
}),
|
||||
getModelDownloadProxyUrl: () => proxy.url,
|
||||
idleTimeoutMs: 1000,
|
||||
})
|
||||
|
||||
await client.status('session-a')
|
||||
proxy.url = 'http://127.0.0.1:7890'
|
||||
await run(client)
|
||||
|
||||
assert.equal(instances.length, 2, action)
|
||||
assert.equal(instances[0].closed, true, action)
|
||||
assert.deepEqual(instances[0].requests, [{ method: 'status', args: ['session-a'] }], action)
|
||||
assert.deepEqual(instances[1].requests, [expectedRequest], action)
|
||||
}
|
||||
})
|
||||
|
||||
test('worker client does not restart api worker before build when proxy changes', async () => {
|
||||
const instances: FakeTransport[] = []
|
||||
const proxy = { url: undefined as string | undefined }
|
||||
const configStore = makeConfigStore()
|
||||
configStore.set({
|
||||
version: 1,
|
||||
enabled: true,
|
||||
mode: 'api',
|
||||
local: { modelId: 'model-a' },
|
||||
api: { baseUrl: 'https://api.example.com/v1', model: 'embed-1' },
|
||||
})
|
||||
const client = createSemanticIndexWorkerClient({
|
||||
configStore,
|
||||
transportFactory: makeFactory(instances, (method) => {
|
||||
if (method === 'status') return null
|
||||
if (method === 'build') return undefined
|
||||
throw new Error(`unexpected method: ${method}`)
|
||||
}),
|
||||
getModelDownloadProxyUrl: () => proxy.url,
|
||||
idleTimeoutMs: 1000,
|
||||
})
|
||||
|
||||
await client.status('session-a')
|
||||
proxy.url = 'http://127.0.0.1:7890'
|
||||
await client.build('session-a')
|
||||
|
||||
assert.equal(instances.length, 1)
|
||||
assert.equal(instances[0].closed, false)
|
||||
assert.deepEqual(instances[0].requests, [
|
||||
{ method: 'status', args: ['session-a'] },
|
||||
{ method: 'build', args: ['session-a'] },
|
||||
])
|
||||
})
|
||||
|
||||
test('worker client clears tracked active builds when semantic indexing is disabled', async () => {
|
||||
const instances: FakeTransport[] = []
|
||||
const timers: Array<() => void> = []
|
||||
|
||||
@@ -23,7 +23,9 @@ export interface SemanticIndexWorkerTransport {
|
||||
close(): void | Promise<void>
|
||||
}
|
||||
|
||||
export type SemanticIndexWorkerTransportFactory = () => SemanticIndexWorkerTransport
|
||||
type MaybePromise<T> = T | Promise<T>
|
||||
|
||||
export type SemanticIndexWorkerTransportFactory = (modelDownloadProxyUrl?: string) => SemanticIndexWorkerTransport
|
||||
|
||||
interface WorkerClientTimers {
|
||||
setTimeout(callback: () => void, ms: number): unknown
|
||||
@@ -35,6 +37,7 @@ export interface SemanticIndexWorkerClientOptions {
|
||||
transportFactory: SemanticIndexWorkerTransportFactory
|
||||
idleTimeoutMs?: number
|
||||
timers?: WorkerClientTimers
|
||||
getModelDownloadProxyUrl?: () => MaybePromise<string | undefined>
|
||||
resolveApiKey?: (provider: string, authProfile?: string) => string
|
||||
writeAuthProfile?: (name: string, profile: AuthProfile) => void
|
||||
}
|
||||
@@ -46,6 +49,7 @@ export interface SemanticIndexWorkerRuntimeClientOptions {
|
||||
sqliteVecLoadablePath?: string
|
||||
workerEntryUrl?: string | URL
|
||||
idleTimeoutMs?: number
|
||||
getModelDownloadProxyUrl?: () => MaybePromise<string | undefined>
|
||||
resolveApiKey?: (provider: string, authProfile?: string) => string
|
||||
writeAuthProfile?: (name: string, profile: AuthProfile) => void
|
||||
}
|
||||
@@ -63,10 +67,12 @@ export class SemanticIndexWorkerClient implements SemanticIndexRuntime {
|
||||
private idleTimer: unknown = null
|
||||
private activeBuildSessionIds = new Set<string>()
|
||||
private hasUnknownActiveBuild = false
|
||||
private activeModelDownloadProxyUrl: string | undefined
|
||||
private readonly configStore: SemanticIndexConfigStore
|
||||
private readonly transportFactory: SemanticIndexWorkerTransportFactory
|
||||
private readonly idleTimeoutMs: number
|
||||
private readonly timers: WorkerClientTimers
|
||||
private readonly getModelDownloadProxyUrl: () => MaybePromise<string | undefined>
|
||||
private readonly resolveApiKey: (provider: string, authProfile?: string) => string
|
||||
private readonly writeAuthProfile?: (name: string, profile: AuthProfile) => void
|
||||
|
||||
@@ -75,6 +81,7 @@ export class SemanticIndexWorkerClient implements SemanticIndexRuntime {
|
||||
this.transportFactory = options.transportFactory
|
||||
this.idleTimeoutMs = options.idleTimeoutMs ?? DEFAULT_IDLE_TIMEOUT_MS
|
||||
this.timers = options.timers ?? defaultTimers
|
||||
this.getModelDownloadProxyUrl = options.getModelDownloadProxyUrl ?? (() => undefined)
|
||||
this.resolveApiKey = options.resolveApiKey ?? defaultResolveApiKey
|
||||
this.writeAuthProfile = options.writeAuthProfile
|
||||
}
|
||||
@@ -93,6 +100,7 @@ export class SemanticIndexWorkerClient implements SemanticIndexRuntime {
|
||||
// 其余情况(disabled / api 模式)无需唤醒 worker,直接返回本地保存结果。
|
||||
const needsWorker = saved.enabled && saved.mode === 'local' && this.configStore.isConfigured()
|
||||
if (!this.transport && !needsWorker) return saved
|
||||
if (needsWorker) await this.closeTransportForLocalModelProxyChangeIfNeeded()
|
||||
return await this.call<SemanticIndexConfig>('setConfig', [saved])
|
||||
}
|
||||
|
||||
@@ -118,6 +126,7 @@ export class SemanticIndexWorkerClient implements SemanticIndexRuntime {
|
||||
}
|
||||
|
||||
async build(sessionId: string): Promise<void> {
|
||||
await this.closeTransportForLocalModelProxyChangeIfNeeded()
|
||||
this.activeBuildSessionIds.add(sessionId)
|
||||
try {
|
||||
await this.call('build', [sessionId])
|
||||
@@ -136,6 +145,7 @@ export class SemanticIndexWorkerClient implements SemanticIndexRuntime {
|
||||
}
|
||||
|
||||
async rebuild(sessionId: string): Promise<void> {
|
||||
await this.closeTransportForLocalModelProxyChangeIfNeeded()
|
||||
this.activeBuildSessionIds.add(sessionId)
|
||||
try {
|
||||
await this.call('rebuild', [sessionId])
|
||||
@@ -146,6 +156,7 @@ export class SemanticIndexWorkerClient implements SemanticIndexRuntime {
|
||||
}
|
||||
|
||||
async buildAllPending(): Promise<void> {
|
||||
await this.closeTransportForLocalModelProxyChangeIfNeeded()
|
||||
this.hasUnknownActiveBuild = true
|
||||
try {
|
||||
await this.call('buildAllPending', [])
|
||||
@@ -187,20 +198,22 @@ export class SemanticIndexWorkerClient implements SemanticIndexRuntime {
|
||||
return await this.callProbe<boolean>('canSearch', [sessionId], false)
|
||||
}
|
||||
|
||||
search(
|
||||
async search(
|
||||
sessionId: string,
|
||||
query: string,
|
||||
options?: { finalTopK?: number; timeRangeMs?: { startTs?: number; endTs?: number } }
|
||||
): Promise<SemanticSearchResult> {
|
||||
return this.call('search', [sessionId, query, options])
|
||||
await this.closeTransportForLocalModelProxyChangeIfNeeded()
|
||||
return await this.call('search', [sessionId, query, options])
|
||||
}
|
||||
|
||||
searchForTool(
|
||||
async searchForTool(
|
||||
sessionId: string,
|
||||
query: string,
|
||||
options?: SemanticSearchToolOptions
|
||||
): Promise<SemanticSearchToolResult> {
|
||||
return this.call('searchForTool', [sessionId, query, options])
|
||||
await this.closeTransportForLocalModelProxyChangeIfNeeded()
|
||||
return await this.call('searchForTool', [sessionId, query, options])
|
||||
}
|
||||
|
||||
cleanupUnused(): Promise<{ cleaned: number }> {
|
||||
@@ -217,7 +230,7 @@ export class SemanticIndexWorkerClient implements SemanticIndexRuntime {
|
||||
}
|
||||
|
||||
private async call<T>(method: string, args: unknown[], options?: { scheduleIdle?: boolean }): Promise<T> {
|
||||
const transport = this.ensureTransport()
|
||||
const transport = await this.ensureTransport()
|
||||
this.pendingRequests++
|
||||
this.clearIdleTimer()
|
||||
try {
|
||||
@@ -242,13 +255,31 @@ export class SemanticIndexWorkerClient implements SemanticIndexRuntime {
|
||||
}
|
||||
}
|
||||
|
||||
private ensureTransport(): SemanticIndexWorkerTransport {
|
||||
private async ensureTransport(): Promise<SemanticIndexWorkerTransport> {
|
||||
if (!this.transport) {
|
||||
this.transport = this.transportFactory()
|
||||
const proxyUrl = await this.currentModelDownloadProxyUrl()
|
||||
if (!this.transport) {
|
||||
this.activeModelDownloadProxyUrl = proxyUrl
|
||||
this.transport = this.transportFactory(proxyUrl)
|
||||
}
|
||||
}
|
||||
return this.transport
|
||||
}
|
||||
|
||||
private async currentModelDownloadProxyUrl(): Promise<string | undefined> {
|
||||
const proxyUrl = (await this.getModelDownloadProxyUrl())?.trim()
|
||||
return proxyUrl || undefined
|
||||
}
|
||||
|
||||
private async closeTransportForLocalModelProxyChangeIfNeeded(): Promise<void> {
|
||||
if (!this.transport) return
|
||||
const config = this.configStore.get()
|
||||
if (!this.configStore.canRun() || config.mode !== 'local') return
|
||||
if (this.hasUnknownActiveBuild || this.activeBuildSessionIds.size > 0) return
|
||||
if ((await this.currentModelDownloadProxyUrl()) === this.activeModelDownloadProxyUrl) return
|
||||
await this.closeTransport()
|
||||
}
|
||||
|
||||
private canRunFromLocalConfig(): boolean {
|
||||
const config = this.configStore.get()
|
||||
if (!this.configStore.canRun()) return false
|
||||
@@ -311,6 +342,7 @@ export class SemanticIndexWorkerClient implements SemanticIndexRuntime {
|
||||
private async closeTransport(): Promise<void> {
|
||||
const transport = this.transport
|
||||
this.transport = null
|
||||
this.activeModelDownloadProxyUrl = undefined
|
||||
this.clearActiveBuildTracking()
|
||||
if (transport) await transport.close()
|
||||
}
|
||||
@@ -340,7 +372,8 @@ export function createSemanticIndexWorkerRuntimeClient(
|
||||
idleTimeoutMs: options.idleTimeoutMs,
|
||||
resolveApiKey: options.resolveApiKey,
|
||||
writeAuthProfile: options.writeAuthProfile,
|
||||
transportFactory: () =>
|
||||
getModelDownloadProxyUrl: options.getModelDownloadProxyUrl,
|
||||
transportFactory: (modelDownloadProxyUrl) =>
|
||||
createSemanticIndexWorkerThreadTransport({
|
||||
workerEntryUrl: options.workerEntryUrl,
|
||||
startup: {
|
||||
@@ -348,6 +381,7 @@ export function createSemanticIndexWorkerRuntimeClient(
|
||||
runtime: options.runtime,
|
||||
nativeBinding: options.nativeBinding,
|
||||
sqliteVecLoadablePath: options.sqliteVecLoadablePath,
|
||||
modelDownloadProxyUrl,
|
||||
},
|
||||
}),
|
||||
})
|
||||
|
||||
@@ -1,7 +1,12 @@
|
||||
import assert from 'node:assert/strict'
|
||||
import test from 'node:test'
|
||||
import { createSemanticIndexWorkerRuntime, type SemanticIndexWorkerServiceFactory } from './worker-runtime'
|
||||
import {
|
||||
createSemanticIndexWorkerRuntime,
|
||||
createSemanticIndexWorkerServiceFactory,
|
||||
type SemanticIndexWorkerServiceFactory,
|
||||
} from './worker-runtime'
|
||||
import type { SemanticIndexRuntime } from './runtime'
|
||||
import type { SemanticIndexWorkerStartupOptions } from './worker-runtime'
|
||||
|
||||
class FakeSemanticIndexRuntime implements Partial<SemanticIndexRuntime> {
|
||||
statusCalls: string[] = []
|
||||
@@ -21,6 +26,24 @@ class FakeSemanticIndexRuntime implements Partial<SemanticIndexRuntime> {
|
||||
}
|
||||
}
|
||||
|
||||
function makeStartupOptions(logsDir: string): SemanticIndexWorkerStartupOptions {
|
||||
return {
|
||||
paths: {
|
||||
systemDir: '/tmp/system',
|
||||
userDataDir: '/tmp/data',
|
||||
databaseDir: '/tmp/data/databases',
|
||||
vectorDir: '/tmp/data/vector',
|
||||
aiDataDir: '/tmp/system/ai',
|
||||
settingsDir: '/tmp/system/settings',
|
||||
cacheDir: '/tmp/system/cache',
|
||||
tempDir: '/tmp/system/temp',
|
||||
logsDir,
|
||||
downloadsDir: '/tmp/downloads',
|
||||
},
|
||||
runtime: { version: '0.0.0-test', kind: 'cli' },
|
||||
}
|
||||
}
|
||||
|
||||
test('worker runtime lazily creates service and forwards RPC calls', async () => {
|
||||
const services: FakeSemanticIndexRuntime[] = []
|
||||
const factory: SemanticIndexWorkerServiceFactory = () => {
|
||||
@@ -42,3 +65,13 @@ test('worker runtime lazily creates service and forwards RPC calls', async () =>
|
||||
|
||||
assert.equal(services[0].closed, true)
|
||||
})
|
||||
|
||||
test('worker service factory initializes app logger with worker logs dir', () => {
|
||||
const calls: string[] = []
|
||||
|
||||
createSemanticIndexWorkerServiceFactory(makeStartupOptions('/tmp/system/logs'), {
|
||||
initLogger: (logsDir) => calls.push(logsDir),
|
||||
})
|
||||
|
||||
assert.deepEqual(calls, ['/tmp/system/logs'])
|
||||
})
|
||||
@@ -1,5 +1,6 @@
|
||||
import type { RuntimeIdentity } from '../data-dir-compat'
|
||||
import { DatabaseManager } from '../database-manager'
|
||||
import { initAppLogger } from '../logging/app-logger'
|
||||
import { createDatabaseManagerAdapter } from '../services'
|
||||
import { createSemanticIndexService } from './service'
|
||||
import type { LoadSqliteVec } from './store'
|
||||
@@ -11,10 +12,15 @@ export interface SemanticIndexWorkerStartupOptions {
|
||||
runtime: RuntimeIdentity
|
||||
nativeBinding?: string
|
||||
sqliteVecLoadablePath?: string
|
||||
modelDownloadProxyUrl?: string
|
||||
}
|
||||
|
||||
export type SemanticIndexWorkerServiceFactory = () => SemanticIndexRuntime
|
||||
|
||||
export interface SemanticIndexWorkerServiceFactoryDeps {
|
||||
initLogger?: (logsDir: string) => void
|
||||
}
|
||||
|
||||
export interface SemanticIndexWorkerRuntimeOptions {
|
||||
serviceFactory: SemanticIndexWorkerServiceFactory
|
||||
}
|
||||
@@ -61,8 +67,12 @@ export function createSemanticIndexWorkerRuntime(
|
||||
}
|
||||
|
||||
export function createSemanticIndexWorkerServiceFactory(
|
||||
options: SemanticIndexWorkerStartupOptions
|
||||
options: SemanticIndexWorkerStartupOptions,
|
||||
deps: SemanticIndexWorkerServiceFactoryDeps = {}
|
||||
): SemanticIndexWorkerServiceFactory {
|
||||
const initializeLogger = deps.initLogger ?? initAppLogger
|
||||
initializeLogger(options.paths.logsDir)
|
||||
|
||||
return () => {
|
||||
const pathProvider = new StaticPathProvider(options.paths)
|
||||
const dbManager = new DatabaseManager(pathProvider, {
|
||||
@@ -79,6 +89,7 @@ export function createSemanticIndexWorkerServiceFactory(
|
||||
sessionAdapter,
|
||||
nativeBinding: options.nativeBinding,
|
||||
loadSqliteVec,
|
||||
modelDownloadProxyUrl: options.modelDownloadProxyUrl,
|
||||
})
|
||||
}
|
||||
}
|
||||
Generated
+3
@@ -346,6 +346,9 @@ importers:
|
||||
stream-json:
|
||||
specifier: ^1.9.1
|
||||
version: 1.9.1
|
||||
undici:
|
||||
specifier: ^6.25.0
|
||||
version: 6.25.0
|
||||
yauzl:
|
||||
specifier: ^3.4.0
|
||||
version: 3.4.0
|
||||
|
||||
Reference in new issue
Block a user