fix(semantic-index): stabilize embedding warmup (#261)

* fix(semantic-index): stabilize embedding warmup

Improve warmup job reliability and prevent raw API response body from
reaching app.log. Status code alone is sufficient for diagnosing failures;
stripping the response detail eliminates the risk of chat content being
written to the log via OpenAI-compatible services that echo submitted input.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* fix(semantic-index): clear preload error after warmup retry

When local model preload fails, a later warmup can successfully recreate the extractor through embedDocuments but leave modelPreloadStatus stuck on error.

Mark the local model ready after a completed warmup writes chunks, and cover the retry path with a regression test.

* fix(semantic-index): guard warmup model status updates

Capture the model configuration used by a warmup job before it starts, and only clear local preload errors when the current local model still matches that job.

Add a regression test for switching local models while an older warmup is still completing.

---------

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>

packages/node-runtime/src/semantic-index/embedding/api.ts

@@ -10,8 +10,8 @@
 
 import type { EmbeddingProvider } from './types'
 
-/** API embedding 默认一次提交 32 个 chunk，保持 concurrency=1 以降低 429 风险。 */
-export const API_DOCUMENT_BATCH_SIZE = 32
+/** API embedding 默认一次提交 8 个 chunk，兼容更多 OpenAI-compatible 服务的批量上限。 */
+export const API_DOCUMENT_BATCH_SIZE = 8
 
 export interface OpenAICompatibleConfig {
   baseUrl: string
@@ -74,8 +74,8 @@ export class OpenAICompatibleEmbeddingProvider implements EmbeddingProvider {
     })
 
     if (!response.ok) {
-      const detail = await response.text().catch(() => '')
-      throw new Error(`Embedding API request failed: ${response.status} ${detail}`.trim())
+      await response.text().catch(() => '') // consume body to avoid connection leak
+      throw new Error(`Embedding API request failed: ${response.status}`)
     }
 
     const payload = (await response.json()) as EmbeddingResponse

packages/node-runtime/src/semantic-index/embedding/local.ts

@@ -15,6 +15,7 @@ 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:'])
+export const LOCAL_ONNX_SESSION_OPTIONS = { intraOpNumThreads: 1, interOpNumThreads: 1 } as const
 
 /** 一次特征抽取调用：输入文本数组，返回每条文本的向量（number[][]） */
 export type FeatureExtractFn = (
@@ -28,6 +29,7 @@ export type LocalPipelineFactory = (params: {
   dtype?: 'fp32' | 'q8'
   cacheDir?: string
   modelDownloadProxyUrl?: string
+  sessionOptions?: typeof LOCAL_ONNX_SESSION_OPTIONS
 }) => Promise<FeatureExtractFn>
 
 export async function createProxyFetch(proxyUrl: string): Promise<typeof fetch> {
@@ -54,7 +56,10 @@ const defaultPipelineFactory: LocalPipelineFactory = async ({ modelId, dtype, ca
   if (modelDownloadProxyUrl) {
     transformers.env.fetch = await createProxyFetch(modelDownloadProxyUrl)
   }
-  const extractor = await transformers.pipeline('feature-extraction', modelId, dtype ? { dtype } : undefined)
+  const extractor = await transformers.pipeline('feature-extraction', modelId, {
+    ...(dtype ? { dtype } : {}),
+    session_options: LOCAL_ONNX_SESSION_OPTIONS,
+  })
   return async (texts, options) => {
     const out = await extractor(texts, { pooling: options.pooling, normalize: options.normalize })
     return out.tolist() as number[][]
@@ -94,6 +99,10 @@ export class LocalEmbeddingProvider implements EmbeddingProvider {
         dtype: this.profile.dtype,
         cacheDir: this.options.cacheDir,
         modelDownloadProxyUrl: this.options.modelDownloadProxyUrl,
+        sessionOptions: LOCAL_ONNX_SESSION_OPTIONS,
+      }).catch((error) => {
+        this.extractorPromise = null
+        throw error
       })
     }
     return this.extractorPromise

packages/node-runtime/src/semantic-index/embedding/providers.test.ts

@@ -27,6 +27,20 @@ test('Qwen3 local provider embeds one text per call (batch=1)', async () => {
   assert.ok(calls.every((c) => c.length === 1))
 })
 
+test('local provider limits ONNX runtime threads by default', async () => {
+  let capturedSessionOptions: unknown
+  const provider = new LocalEmbeddingProvider(QWEN3_PROFILE, {
+    pipelineFactory: async (params) => {
+      capturedSessionOptions = params.sessionOptions
+      return async (texts) => texts.map(() => Array.from({ length: QWEN3_PROFILE.dim }, () => 0.1))
+    },
+  })
+
+  await provider.preload()
+
+  assert.deepEqual(capturedSessionOptions, { intraOpNumThreads: 1, interOpNumThreads: 1 })
+})
+
 test('local provider without maxBatchSize embeds all texts in a single call', async () => {
   const calls: string[][] = []
   const provider = new LocalEmbeddingProvider(NO_BATCH_PROFILE, { pipelineFactory: makeFakeFactory(8, calls) })
@@ -61,6 +75,22 @@ test('local provider throws when returned dim mismatches profile', async () => {
   await assert.rejects(() => provider.embedDocuments(['a']), /dim/i)
 })
 
+test('local provider retries pipeline creation after a failed preload', async () => {
+  let attempts = 0
+  const provider = new LocalEmbeddingProvider(QWEN3_PROFILE, {
+    pipelineFactory: async () => {
+      attempts++
+      if (attempts === 1) throw new Error('temporary network failure')
+      return async (texts) => texts.map(() => Array.from({ length: QWEN3_PROFILE.dim }, () => 0.1))
+    },
+  })
+
+  await assert.rejects(() => provider.preload(), /temporary network failure/)
+  await provider.preload()
+
+  assert.equal(attempts, 2)
+})
+
 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/)
 })
@@ -86,7 +116,7 @@ test('API provider posts OpenAI-compatible request and parses ordered embeddings
     { fetchFn }
   )
 
-  assert.equal(provider.documentBatchSize, 32)
+  assert.equal(provider.documentBatchSize, 8)
   const vectors = await provider.embedDocuments(['first', 'second'])
   assert.equal(captured!.url, 'https://api.example.com/v1/embeddings')
   assert.equal(captured!.headers.Authorization, 'Bearer sk-test')

packages/node-runtime/src/semantic-index/service.test.ts

@@ -15,7 +15,7 @@ import {
 import { SemanticIndexConfigStore } from './config'
 import { SemanticIndexStateStore } from './session-state-store'
 import { computeDbPathHash } from './chunker-config'
-import { QWEN3_PROFILE } from './embedding/profiles'
+import { BGE_BASE_PROFILE, QWEN3_PROFILE } from './embedding/profiles'
 import type { SessionRuntimeAdapter } from '../services/adapters'
 import type { LocalPipelineFactory } from './embedding/local'
 
@@ -24,7 +24,12 @@ const SESSION_ID = 'sess1'
 // 暴露「毫秒被当作秒再 *1000」导致五位数年份（如 56150）的回归。
 const BASE_TS_SECONDS = 1_700_000_000
 
-function setup(opts?: { embedDelayMs?: number }) {
+function setup(opts?: {
+  embedDelayMs?: number
+  failFirstPipelineCreation?: boolean
+  failPipelineCreationForModelIds?: ReadonlySet<string>
+  onEmbedBatchStarted?: () => void
+}) {
   const baseDir = fs.existsSync('/private/tmp') ? '/private/tmp' : os.tmpdir()
   const dir = fs.mkdtempSync(path.join(baseDir, 'chatlab-si-svc-'))
   const chatDbPath = path.join(dir, `${SESSION_ID}.db`)
@@ -61,14 +66,25 @@ function setup(opts?: { embedDelayMs?: number }) {
   // 本地 pipeline 工厂：返回 Qwen3 维度向量，按文本长度给一点变化，避免零向量
   // embedTextCount 统计累计嵌入文本数，用于区分 rebuild(重新嵌入) 与 build 续跑(跳过不嵌入)
   let embedTextCount = 0
-  const localPipelineFactory: LocalPipelineFactory = async () => async (texts) => {
-    embedTextCount += texts.length
-    if (opts?.embedDelayMs) await new Promise((r) => setTimeout(r, opts.embedDelayMs))
-    return texts.map((t) => {
-      const v = new Array(QWEN3_PROFILE.dim).fill(0)
-      v[t.length % QWEN3_PROFILE.dim] = 1
-      return v
-    })
+  let pipelineCreateAttempts = 0
+  const localPipelineFactory: LocalPipelineFactory = async ({ modelId }) => {
+    pipelineCreateAttempts++
+    if (opts?.failFirstPipelineCreation && pipelineCreateAttempts === 1) {
+      throw new Error('temporary preload failure')
+    }
+    if (opts?.failPipelineCreationForModelIds?.has(modelId)) {
+      throw new Error(`temporary preload failure for ${modelId}`)
+    }
+    return async (texts) => {
+      opts?.onEmbedBatchStarted?.()
+      embedTextCount += texts.length
+      if (opts?.embedDelayMs) await new Promise((r) => setTimeout(r, opts.embedDelayMs))
+      return texts.map((t) => {
+        const v = new Array(QWEN3_PROFILE.dim).fill(0)
+        v[t.length % QWEN3_PROFILE.dim] = 1
+        return v
+      })
+    }
   }
 
   // 内存 auth profile 存储：避免测试写真实 ~/.chatlab/auth-profiles.json
@@ -96,6 +112,18 @@ async function enableAndBuild(service: SemanticIndexService): Promise<void> {
   await service.whenIdle()
 }
 
+async function waitForModelStatus(
+  service: SemanticIndexService,
+  expected: ReturnType<SemanticIndexService['getModelStatus']>
+): Promise<void> {
+  const deadline = Date.now() + 1000
+  while (Date.now() < deadline) {
+    if (service.getModelStatus() === expected) return
+    await new Promise((resolve) => setTimeout(resolve, 10))
+  }
+  assert.equal(service.getModelStatus(), expected)
+}
+
 test('explicit build indexes an enabled session to completion and reports status', async () => {
   const { service, db } = setup()
   await enableAndBuild(service)
@@ -139,6 +167,42 @@ test('search returns evidence blocks for an enabled completed index', async () =
   db.close()
 })
 
+test('successful warmup retry clears a previous local model preload error', async () => {
+  const { service, db } = setup({ failFirstPipelineCreation: true })
+
+  await waitForModelStatus(service, 'error')
+  await enableAndBuild(service)
+
+  assert.equal(service.status(SESSION_ID)!.indexStatus, 'completed')
+  assert.equal(service.getModelStatus(), 'ready')
+  service.close()
+  db.close()
+})
+
+test('stale local warmup completion does not mark current local model ready', async () => {
+  let resolveFirstBatchStarted: () => void = () => {}
+  const firstBatchStarted = new Promise<void>((resolve) => {
+    resolveFirstBatchStarted = resolve
+  })
+  const { service, db } = setup({
+    embedDelayMs: 50,
+    failPipelineCreationForModelIds: new Set([BGE_BASE_PROFILE.modelId]),
+    onEmbedBatchStarted: resolveFirstBatchStarted,
+  })
+
+  service.enable(SESSION_ID)
+  service.build(SESSION_ID)
+  await firstBatchStarted
+
+  service.setConfig({ version: 1, mode: 'local', local: { modelId: BGE_BASE_PROFILE.modelId }, api: null })
+  await waitForModelStatus(service, 'error')
+  await service.whenIdle()
+
+  assert.equal(service.getModelStatus(), 'error')
+  service.close()
+  db.close()
+})
+
 test('changing model identity marks needsRebuild and blocks search until rebuilt', async () => {
   const { service, db } = setup()
   await enableAndBuild(service)

packages/node-runtime/src/semantic-index/service.ts

@@ -822,16 +822,41 @@ export class SemanticIndexService {
       })
     }
 
+    const jobConfig = this.configStore.get()
+    const jobModelId = this.currentModelId()
+    const jobEmbedder = this.getEmbedder()
     const source = createChatDbMessageSource(db, this.resolveSource(db))
-    await runWarmup({
+    const startedAt = Date.now()
+    appLogger.info('semantic-index', 'warmup job started', { type: job.type, sessionId, dbPathHash: job.dbPathHash })
+    const result = await runWarmup({
       dbPathHash: job.dbPathHash,
-      modelId: this.currentModelId(),
-      embedder: this.getEmbedder(),
+      modelId: jobModelId,
+      embedder: jobEmbedder,
       store: this.store,
       stateStore: this.stateStore,
       source,
       checkStop,
     })
+    // warmup 已成功写入 chunk，说明本地 extractor 通过重试加载成功；清除此前 preload 失败状态。
+    const currentConfig = this.configStore.get()
+    if (
+      result.status === 'completed' &&
+      result.chunksWritten > 0 &&
+      jobConfig.mode === 'local' &&
+      currentConfig.mode === 'local' &&
+      currentConfig.local.modelId === jobConfig.local.modelId
+    ) {
+      this.modelPreloadStatus = 'ready'
+    }
+    appLogger.info('semantic-index', 'warmup job finished', {
+      type: job.type,
+      sessionId,
+      dbPathHash: job.dbPathHash,
+      status: result.status,
+      chunksWritten: result.chunksWritten,
+      elapsedMs: Date.now() - startedAt,
+      error: result.error,
+    })
   }
 
   /** 等待队列清空（优雅停止 / 测试用） */