Function runWarmup

packages/node-runtime/src/semantic-index/warmup/runner.ts:60–186

(options: WarmupRunnerOptions)

}

export async function runWarmup(options: WarmupRunnerOptions): Promise<WarmupResult> {
  const { dbPathHash, modelId, embedder, store, stateStore, source, config, checkStop } = options

  let chunksWritten = 0
  try {
    const messages = source.readAllMessages()
    const total = source.countMessages()
    const savedState = stateStore.getState(dbPathHash)
    stateStore.updateProgress(dbPathHash, { indexStatus: 'running', totalMessages: total, error: null })

    // 消息 id -> 流位置，用于进度计数（消息按 ts 排序，id 未必单调）
    const streamIndexById = new Map<number, number>()
    messages.forEach((m, i) => streamIndexById.set(m.id, i))

    const resumeMessageId = savedState?.lastIndexedMessageId ?? null
    const rawResumeIndex = resumeMessageId !== null ? (streamIndexById.get(resumeMessageId) ?? -1) : -1

    // Detect non-append-only additions: if the cursor's new stream position doesn't match the
    // saved indexed count, older messages were backfilled before the cursor. Clear vectors and
    // re-index from scratch to avoid silently missing the backfilled messages.
    const isNonAppendOnly = rawResumeIndex >= 0 && rawResumeIndex + 1 !== (savedState?.indexedMessages ?? 0)
    if (isNonAppendOnly) {
      store.deleteByDbPathHash(dbPathHash)
    }
    const { chunks, chunkerConfigHash } = chunkMessages({ messages, source: source.getSource(), config })

    const chunkRanges = chunks.map((chunk) => ({
      chunk,
      startIndex: streamIndexById.get(chunk.startMessageId) ?? -1,
      endIndex: streamIndexById.get(chunk.endMessageId) ?? -1,
    }))

    let resumeIndex = isNonAppendOnly ? -1 : rawResumeIndex
    const hasAppendedMessages = !isNonAppendOnly && rawResumeIndex >= 0 && total > (savedState?.totalMessages ?? 0)
    if (hasAppendedMessages) {
      const cursorRange =
        chunkRanges.find((range) => range.startIndex <= rawResumeIndex && rawResumeIndex <= range.endIndex) ??
        chunkRanges.filter((range) => range.endIndex <= rawResumeIndex).sort((a, b) => b.endIndex - a.endIndex)[0]

      if (cursorRange) {
        const parentRanges = chunkRanges.filter((range) => range.chunk.parentId === cursorRange.chunk.parentId)
        const rewindIndex = Math.min(...parentRanges.map((range) => range.startIndex).filter((index) => index >= 0))
        const rewindMessage = messages[rewindIndex]
        if (rewindMessage) {
          store.deleteByModelFromPosition({
            dbPathHash,
            modelId,
            startTs: rewindMessage.ts,
            startMessageId: rewindMessage.id,
          })
          resumeIndex = rewindIndex - 1
        }
      }
    }

    // 续跑时统计已写入 chunk 数，保证 chunkCount 连续
    let storedChunkCount = store.countChunks(dbPathHash, modelId)
    if (resumeIndex < 0) storedChunkCount = 0