opensource/kilocode
3
0
Fork 0
mirror of https://github.com/Kilo-Org/kilocode.git synced 2026-08-30 17:14:40 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(cli): refine shared index concurrency

Browse Source
This commit is contained in:
marius-kilocode
2026-06-16 15:56:44 +02:00
parent c6b335c6e5
commit 42627729b4
18 changed files with 513 additions and 212 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/kilo-indexing/src/indexing/cache-manager.ts
+7
View File
@@ -91,4 +91,11 @@ export class CacheManager implements ICacheManager {
const entries = Object.entries(this.fileHashes).sort(([left], [right]) => left.localeCompare(right))
return createHash("sha256").update(JSON.stringify(entries)).digest("hex")
}
async stamp(): Promise<string | undefined> {
return fs
.stat(this.cachePath)
.then((value) => `${value.mtimeMs}:${value.ctimeMs}:${value.size}`)
.catch(() => undefined)
}
}
packages/kilo-indexing/src/indexing/interfaces/vector-store.ts
+1
View File
@@ -96,6 +96,7 @@ export interface VectorStoreSearchResult {
export interface Payload {
filePath: string
fileHash?: string
codeChunk: string
startLine: number
endLine: number
packages/kilo-indexing/src/indexing/manager.ts
+81 -39
View File
@@ -16,6 +16,15 @@ import { sanitizeErrorMessage } from "./shared/validation-helpers"
import { WorktreeOverlay } from "./worktree-overlay"
const log = Log.create({ service: "indexing-manager" })
const BASELINE_CHECK_INTERVAL = 1_000
const BASELINE_SIGNATURE_INTERVAL = 30_000
type Baseline = {
store?: IVectorStore
signature: string
stamp?: string
overlay?: WorktreeOverlay
}
/**
* RATIONALE: Removed the static singleton Map and vscode.ExtensionContext.
@@ -33,6 +42,9 @@ export class CodeIndexManager {
private _cacheManager: CacheManager | undefined
private _baselineStore: IVectorStore | undefined
private _baselineSignature: string | undefined
private _baselineStamp: string | undefined
private _baselineChecked = 0
private _baselineSigned = 0
private _baselineRefresh: Promise<void> | undefined
private _overlay: WorktreeOverlay | undefined
private _isRecoveringFromError = false
@@ -449,18 +461,36 @@ export class CodeIndexManager {
private async refreshBaseline(): Promise<void> {
if (!this.baselinePath || this._disposed) return
if (this._baselineRefresh) return this._baselineRefresh
const now = Date.now()
if (now - this._baselineChecked < BASELINE_CHECK_INTERVAL) return
this._baselineChecked = now
const baselinePath = this.baselinePath
const task = (async () => {
const cache = new CacheManager(this.cacheDirectory, baselinePath)
const stamp = await cache.stamp()
const force = now - this._baselineSigned >= BASELINE_SIGNATURE_INTERVAL
if (!force && stamp === this._baselineStamp) return
await cache.initialize()
if (cache.signature() === this._baselineSignature) return
const signature = cache.signature()
this._baselineStamp = stamp
this._baselineSigned = now
if (signature === this._baselineSignature && this._baselineStore) return
const baseline = await this.createBaseline(this._serviceFactory!)
this._baselineStamp = baseline?.stamp ?? stamp
this._baselineSigned = now
if (!baseline?.store) {
if (!this._baselineStore) this._baselineSignature = signature
return
}
log.info("shared indexing baseline changed; rebuilding worktree delta", {
workspacePath: this.workspacePath,
baselinePath,
})
await this._recreateServices()
await this._recreateServices(baseline)
if (this._disposed) return
await this._orchestrator?.startIndexing("background")
})().finally(() => {
@@ -470,11 +500,13 @@ export class CodeIndexManager {
return task
}
private async createBaseline(factory: CodeIndexServiceFactory) {
private async createBaseline(factory: CodeIndexServiceFactory): Promise<Baseline | undefined> {
if (!this.baselinePath) return
const cache = new CacheManager(this.cacheDirectory, this.baselinePath)
await cache.initialize()
const signature = cache.signature()
const stamp = await cache.stamp()
const hashes = new Map<string, string>()
for (const [filePath, hash] of Object.entries(cache.getAllHashes())) {
const rel = path.relative(this.baselinePath, filePath)
@@ -483,47 +515,42 @@ export class CodeIndexManager {
}
const store = factory.createVectorStore(this.baselinePath)
if (!store.openExisting) throw new Error("The configured vector store cannot open a shared baseline")
// Validate compatibility without keeping every worktree baseline connection open.
await store.openExisting()
await store.close?.()
return {
store,
signature: cache.signature(),
overlay: new WorktreeOverlay(this.workspacePath, this.baselinePath, hashes),
try {
if (!store.openExisting) throw new Error("The configured vector store cannot open a shared baseline")
// Validate compatibility without keeping every worktree baseline connection open.
await store.openExisting()
await store.close?.()
return {
store,
signature,
stamp,
overlay: new WorktreeOverlay(this.workspacePath, this.baselinePath, hashes),
}
} catch (err) {
await store.close?.()
log.warn("shared indexing baseline is unavailable; using an independent worktree index", {
workspacePath: this.workspacePath,
baselinePath: this.baselinePath,
err,
})
return { signature, stamp }
}
}
private async _recreateServices(): Promise<void> {
private async _recreateServices(prepared?: Baseline): Promise<void> {
log.info("starting indexing service recreation", { workspacePath: this.workspacePath })
await this._orchestrator?.shutdown?.()
await this._baselineStore?.close?.()
this._orchestrator = undefined
this._baselineStore = undefined
this._searchService = undefined
this._serviceFactory = new CodeIndexServiceFactory(
const factory = new CodeIndexServiceFactory(
this._configManager!,
this.workspacePath,
this._cacheManager!,
this.cacheDirectory,
(event) => this.handleTelemetry(event),
)
const ignoreInstance = await loadIgnore(this.workspacePath)
const config = this._configManager!.getConfig()
const baseline = await this.createBaseline(this._serviceFactory)
this._baselineStore = baseline?.store
this._baselineSignature = baseline?.signature
this._overlay = baseline?.overlay
const { embedder, vectorStore, scanner, fileWatcher } = this._serviceFactory.createServices(
this._cacheManager!,
ignoreInstance,
)
fileWatcher.setOverlay?.(this._overlay)
const baseline = prepared ?? (await this.createBaseline(factory))
const { embedder, vectorStore, scanner, fileWatcher } = factory.createServices(this._cacheManager!, ignoreInstance)
fileWatcher.setOverlay?.(baseline?.overlay)
log.info("created indexing services", {
workspacePath: this.workspacePath,
provider: embedder.embedderInfo.name,
@@ -532,13 +559,12 @@ export class CodeIndexManager {
})
const shouldValidate = embedder && embedder.embedderInfo.name === config.embedderProvider
if (shouldValidate) {
log.info("validating embedder configuration", {
workspacePath: this.workspacePath,
provider: embedder.embedderInfo.name,
})
const validationResult = await this._serviceFactory.validateEmbedder(embedder)
const validationResult = await factory.validateEmbedder(embedder)
if (!validationResult.valid) {
const errorMessage = validationResult.error || "Embedder configuration validation failed"
this._stateManager.setSystemState("Error", errorMessage)
@@ -550,7 +576,7 @@ export class CodeIndexManager {
})
}
this._orchestrator = new CodeIndexOrchestrator(
const orchestrator = new CodeIndexOrchestrator(
this._configManager!,
this._stateManager,
this.workspacePath,
@@ -559,17 +585,33 @@ export class CodeIndexManager {
scanner,
fileWatcher,
(event) => this.handleTelemetry(event),
this._overlay,
baseline?.overlay,
Boolean(this.baselinePath && !baseline?.store),
)
this._searchService = new CodeIndexSearchService(
const search = new CodeIndexSearchService(
this._configManager!,
this._stateManager,
embedder,
vectorStore,
this._baselineStore && this._overlay ? { store: this._baselineStore, overlay: this._overlay } : undefined,
baseline?.store && baseline.overlay ? { store: baseline.store, overlay: baseline.overlay } : undefined,
)
await this._orchestrator?.shutdown?.()
await this._baselineStore?.close?.()
if (this._disposed) {
await orchestrator.shutdown()
await baseline?.store?.close?.()
return
}
this._serviceFactory = factory
this._orchestrator = orchestrator
this._searchService = search
this._baselineStore = baseline?.store
this._baselineSignature = baseline?.signature
this._baselineStamp = baseline?.stamp
this._baselineSigned = Date.now()
this._overlay = baseline?.overlay
this._stateManager.setSystemState("Standby", "")
log.info("indexing services are ready", { workspacePath: this.workspacePath })
}
packages/kilo-indexing/src/indexing/orchestrator.ts
+10 -4
View File
@@ -37,6 +37,7 @@ export class CodeIndexOrchestrator {
private readonly fileWatcher: IFileWatcher,
private readonly onTelemetry?: IndexingTelemetryReporter,
private readonly overlay?: WorktreeOverlay,
private readonly independent = false,
) {}
private getTelemetryMeta(): IndexingTelemetryMeta {
@@ -212,12 +213,17 @@ export class CodeIndexOrchestrator {
baselinePath: this.overlay.baselinePath,
files: this.overlay.baseline.size,
})
} else if (collectionCreated) {
await this.cacheManager.clearCacheFile()
log.info("cleared indexing cache after new collection creation", { workspacePath: this.workspacePath })
}
const hasExistingData = this.overlay ? false : await this.vectorStore.hasIndexedData()
const hasExistingData = this.overlay || this.independent ? false : await this.vectorStore.hasIndexedData()
if (!this.overlay && !hasExistingData) {
if (!collectionCreated) await this.vectorStore.clearCollection()
await this.cacheManager.clearCacheFile()
log.info("cleared indexing cache before full scan", {
workspacePath: this.workspacePath,
collectionCreated,
})
}
log.info("checked vector store indexed data", {
workspacePath: this.workspacePath,
hasExistingData,
packages/kilo-indexing/src/indexing/processors/file-watcher.ts
+4 -1
View File
@@ -472,7 +472,9 @@ export class FileWatcher implements IFileWatcher {
retryCount: this.maxBatchRetries,
})
this.emitError("file-watcher:upsert_retry_exhausted", upsertError, this.maxBatchRetries)
throw new Error(`Failed to upsert batch after ${this.maxBatchRetries} retries: ${upsertError.message}`)
throw new Error(
`Failed to upsert batch after ${this.maxBatchRetries} retries: ${upsertError.message}`,
)
}
this.emitRetry(retryCount, batch.length, upsertError)
await new Promise((resolve) =>
@@ -735,6 +737,7 @@ export class FileWatcher implements IFileWatcher {
vector,
payload: {
filePath: generateRelativeFilePath(normalizedAbsolutePath, this.workspacePath),
fileHash: block.fileHash,
codeChunk: block.content,
startLine: block.start_line,
endLine: block.end_line,
packages/kilo-indexing/src/indexing/processors/scanner.ts
+1
View File
@@ -612,6 +612,7 @@ export class DirectoryScanner implements IDirectoryScanner {
vector,
payload: {
filePath: generateRelativeFilePath(normalizedAbsolutePath, scanWorkspace),
fileHash: block.fileHash,
codeChunk: block.content,
startLine: block.start_line,
endLine: block.end_line,
packages/kilo-indexing/src/indexing/search-service.ts
+4 -7
View File
@@ -58,7 +58,7 @@ export class CodeIndexSearchService {
return search(maxResults)
})()
const base = (async () => {
const checks = new Map<string, Promise<{ valid: boolean; content: string }>>()
const checks = new Map<string, Promise<boolean>>()
const search = async (limit: number): Promise<VectorStoreSearchResult[]> => {
const results = await this.baseline!.store.search(vector, normalizedPrefix, minScore, limit)
const accepted = await Promise.all(
@@ -73,12 +73,9 @@ export class CodeIndexSearchService {
const [baseline, current] = await Promise.all([base, delta])
const merged = new Map<string, VectorStoreSearchResult>()
const key = (result: VectorStoreSearchResult) =>
[
result.payload?.filePath,
result.payload?.startLine,
result.payload?.endLine,
result.payload?.codeChunk,
].join("\0")
[result.payload?.filePath, result.payload?.startLine, result.payload?.endLine, result.payload?.codeChunk].join(
"\0",
)
for (const result of baseline) merged.set(key(result), result)
for (const result of current) {
packages/kilo-indexing/src/indexing/vector-store/lancedb-vector-store.ts
+59 -49
View File
@@ -10,8 +10,20 @@ import type { EmbeddingProfile } from "../embedding-profile"
import { loadLanceDB } from "./lancedb-loader"
const log = Log.create({ service: "lancedb-store" })
let nativeQueue = Promise.resolve()
function native<T>(run: () => Promise<T>): Promise<T> {
const task = nativeQueue.then(run)
nativeQueue = task.then(
() => undefined,
() => undefined,
)
return task
}
const SCHEMA = "2"
const KEY = {
schema: "index_schema",
size: "vector_size",
complete: "indexing_complete",
provider: "embedding_provider",
@@ -74,19 +86,16 @@ export class LanceDBVectorStore implements IVectorStore {
* @returns The LanceDB connection.
*/
private async getDb(): Promise<Connection> {
if (this.db) {
return this.db
}
if (this.db) return this.db
const lancedb = await this.loadLanceDBModule()
return native(async () => {
if (this.db) return this.db
const lancedb = await this.loadLanceDBModule()
// Create parent directory if needed
if (!fs.existsSync(this.dbPath)) {
fs.mkdirSync(this.dbPath, { recursive: true })
}
this.db = await lancedb.connect(this.dbPath)
return this.db as Connection
if (!fs.existsSync(this.dbPath)) fs.mkdirSync(this.dbPath, { recursive: true })
this.db = await lancedb.connect(this.dbPath)
return this.db as Connection
})
}
/**
@@ -102,7 +111,7 @@ export class LanceDBVectorStore implements IVectorStore {
try {
// Try to open existing table
const table = await db.openTable(this.vectorTableName)
const table = await native(() => db.openTable(this.vectorTableName))
this.table = table
return table
} catch (error) {
@@ -121,6 +130,7 @@ export class LanceDBVectorStore implements IVectorStore {
id: "sample",
vector: new Array(this.vectorSize).fill(0),
filePath: "sample",
fileHash: "sample",
codeChunk: "sample",
startLine: 0,
endLine: 0,
@@ -134,6 +144,10 @@ export class LanceDBVectorStore implements IVectorStore {
*/
private _createMetadataData() {
return [
{
key: KEY.schema,
value: SCHEMA,
},
{
key: KEY.size,
value: String(this.vectorSize),
@@ -162,7 +176,7 @@ export class LanceDBVectorStore implements IVectorStore {
* @param db The LanceDB connection.
*/
private async _createVectorTable(db: Connection): Promise<void> {
this.table = await db.createTable(this.vectorTableName, this._createSampleData())
this.table = await native(() => db.createTable(this.vectorTableName, this._createSampleData()))
if (this.table) {
await this.table.delete("id = 'sample'")
}
@@ -173,7 +187,7 @@ export class LanceDBVectorStore implements IVectorStore {
* @param db The LanceDB connection.
*/
private async _createMetadataTable(db: Connection): Promise<void> {
await db.createTable(this.metadataTableName, this._createMetadataData())
await native(() => db.createTable(this.metadataTableName, this._createMetadataData()))
}
/**
@@ -194,15 +208,10 @@ export class LanceDBVectorStore implements IVectorStore {
* @returns The stored vector size, or null if not found.
*/
private async _getStoredVectorSize(db: Connection): Promise<number | null> {
try {
const value = await this._getMetadataValue(db, KEY.size)
if (value === undefined) return null
const dim = this._parseNumber(value)
return dim ?? null
} catch (error) {
log.warn("Failed to read metadata table", { error })
return null
}
const value = await this._getMetadataValue(db, KEY.size)
if (value === undefined) return null
const dim = this._parseNumber(value)
return dim ?? null
}
private isValidMetadataKey(key: string): boolean {
@@ -219,27 +228,22 @@ export class LanceDBVectorStore implements IVectorStore {
if (!this.isValidMetadataKey(key)) {
throw new Error(`Invalid metadata key: ${key}`)
}
const metadataTable = await db.openTable(this.metadataTableName)
const metadataTable = await native(() => db.openTable(this.metadataTableName))
const rows = await metadataTable.query().where(`key = '${key}'`).toArray()
return rows.length > 0 ? rows[0].value : undefined
}
private async _getStoredEmbeddingProfile(db: Connection): Promise<EmbeddingProfile | undefined> {
try {
const provider = await this._getMetadataValue(db, KEY.provider)
const modelId = await this._getMetadataValue(db, KEY.model)
const dimension = await this._getMetadataValue(db, KEY.dimension)
if (typeof provider !== "string" || typeof modelId !== "string") return undefined
const dim = this._parseNumber(dimension)
if (!dim) return undefined
return {
provider: provider as EmbeddingProfile["provider"],
modelId,
dimension: dim,
}
} catch (error) {
log.warn("Failed to read embedding profile metadata", { error })
return undefined
const provider = await this._getMetadataValue(db, KEY.provider)
const modelId = await this._getMetadataValue(db, KEY.model)
const dimension = await this._getMetadataValue(db, KEY.dimension)
if (typeof provider !== "string" || typeof modelId !== "string") return undefined
const dim = this._parseNumber(dimension)
if (!dim) return undefined
return {
provider: provider as EmbeddingProfile["provider"],
modelId,
dimension: dim,
}
}
@@ -265,9 +269,11 @@ export class LanceDBVectorStore implements IVectorStore {
throw new Error("Baseline LanceDB embedding profile does not match the worktree")
}
const schema = await this._getMetadataValue(db, KEY.schema)
if (String(schema) !== SCHEMA) throw new Error("Baseline LanceDB index schema does not match the worktree")
const complete = await this._getMetadataValue(db, KEY.complete)
if (String(complete) !== "true") throw new Error("Baseline LanceDB index is not complete")
this.table = await db.openTable(this.vectorTableName)
this.table = await native(() => db.openTable(this.vectorTableName))
}
async initialize(): Promise<boolean> {
@@ -293,12 +299,13 @@ export class LanceDBVectorStore implements IVectorStore {
return true
}
this.table = await db.openTable(this.vectorTableName)
this.table = await native(() => db.openTable(this.vectorTableName))
const storedVectorSize = metadataTableExists ? await this._getStoredVectorSize(db) : null
const storedSchema = metadataTableExists ? await this._getMetadataValue(db, KEY.schema) : undefined
const pointCount = await this.table.countRows()
if (storedVectorSize === null || storedVectorSize !== this.vectorSize) {
if (String(storedSchema) !== SCHEMA || storedVectorSize === null || storedVectorSize !== this.vectorSize) {
needsRecreation = true
}
@@ -363,6 +370,7 @@ export class LanceDBVectorStore implements IVectorStore {
id: point.id,
vector: point.vector,
filePath: point.payload.filePath,
fileHash: point.payload.fileHash,
codeChunk: point.payload.codeChunk,
startLine: point.payload.startLine,
endLine: point.payload.endLine,
@@ -411,7 +419,7 @@ export class LanceDBVectorStore implements IVectorStore {
if (!payload) {
return false
}
const validKeys = ["filePath", "codeChunk", "startLine", "endLine"]
const validKeys = ["filePath", "fileHash", "codeChunk", "startLine", "endLine"]
const hasValidKeys = validKeys.every((key) => key in payload)
return hasValidKeys
}
@@ -450,6 +458,7 @@ export class LanceDBVectorStore implements IVectorStore {
score: 1 - result._distance, // Convert distance to similarity score
payload: {
filePath: result.filePath,
fileHash: result.fileHash,
codeChunk: result.codeChunk,
startLine: result.startLine,
endLine: result.endLine,
@@ -520,7 +529,7 @@ export class LanceDBVectorStore implements IVectorStore {
const tableNames = await db.tableNames()
if (tableNames.includes(this.metadataTableName)) {
const metadataTable = await db.openTable(this.metadataTableName)
const metadataTable = await native(() => db.openTable(this.metadataTableName))
await metadataTable.delete("true")
}
} catch (metadataError) {
@@ -593,7 +602,7 @@ export class LanceDBVectorStore implements IVectorStore {
})
return false
}
const metadataTable = await db.openTable(this.metadataTableName)
const metadataTable = await native(() => db.openTable(this.metadataTableName))
const metadataResults = await metadataTable.query().where(`key = '${KEY.complete}'`).toArray()
const indexed = metadataResults.length > 0 ? String(metadataResults[0].value) === "true" : false
log.info("LanceDB indexing metadata evaluated", {
@@ -603,8 +612,8 @@ export class LanceDBVectorStore implements IVectorStore {
})
return indexed
} catch (error) {
log.warn("Failed to check if collection has data", { error })
return false
log.error("Failed to check if collection has data", { error })
throw error
}
}
@@ -619,6 +628,7 @@ export class LanceDBVectorStore implements IVectorStore {
}
private async _persistEmbeddingProfile(metadataTable: Table): Promise<void> {
await this._upsertMetadata(metadataTable, KEY.schema, SCHEMA)
await this._upsertMetadata(metadataTable, KEY.provider, this.profile.provider)
await this._upsertMetadata(metadataTable, KEY.model, this.profile.modelId)
await this._upsertMetadata(metadataTable, KEY.dimension, this.profile.dimension)
@@ -632,7 +642,7 @@ export class LanceDBVectorStore implements IVectorStore {
async markIndexingComplete(): Promise<void> {
try {
const db = await this.getDb()
const metadataTable = await db.openTable(this.metadataTableName)
const metadataTable = await native(() => db.openTable(this.metadataTableName))
await this._persistEmbeddingProfile(metadataTable)
await this._upsertMetadata(metadataTable, KEY.complete, "true")
log.info("Marked indexing as complete")
@@ -649,7 +659,7 @@ export class LanceDBVectorStore implements IVectorStore {
async markIndexingIncomplete(): Promise<void> {
try {
const db = await this.getDb()
const metadataTable = await db.openTable(this.metadataTableName)
const metadataTable = await native(() => db.openTable(this.metadataTableName))
await this._persistEmbeddingProfile(metadataTable)
await this._upsertMetadata(metadataTable, KEY.complete, "false")
log.info("Marked indexing as incomplete (in progress)")
packages/kilo-indexing/src/indexing/vector-store/qdrant-client.ts
+14 -16
View File
@@ -9,7 +9,9 @@ import type { EmbeddingProfile } from "../embedding-profile"
const log = Log.create({ service: "qdrant-store" })
const SCHEMA = 2
const KEY = {
schema: "index_schema",
complete: "indexing_complete",
provider: "embedding_provider",
model: "embedding_model_id",
@@ -221,12 +223,12 @@ export class QdrantVectorStore implements IVectorStore {
})
}
private async recreateCollectionForProfile(stored?: EmbeddingProfile): Promise<boolean> {
private async recreateCollectionForCompatibility(stored?: EmbeddingProfile): Promise<boolean> {
const from = stored
? `${stored.provider}:${stored.modelId}:${stored.dimension}`
: "missing embedding metadata on populated collection"
const to = `${this.profile.provider}:${this.profile.modelId}:${this.profile.dimension}`
log.warn(`Collection ${this.collectionName} embedding profile changed (${from} -> ${to}). Recreating collection.`)
log.warn(`Collection ${this.collectionName} is incompatible (${from} -> ${to}). Recreating collection.`)
await this.client.deleteCollection(this.collectionName)
await new Promise((resolve) => setTimeout(resolve, 100))
@@ -258,6 +260,7 @@ export class QdrantVectorStore implements IVectorStore {
if (!profile || !this.isProfileMatch(profile)) {
throw new Error("Baseline Qdrant embedding profile does not match the worktree")
}
if (payload?.[KEY.schema] !== SCHEMA) throw new Error("Baseline Qdrant index schema does not match the worktree")
if (payload?.[KEY.complete] !== true) throw new Error("Baseline Qdrant index is not complete")
}
@@ -299,8 +302,8 @@ export class QdrantVectorStore implements IVectorStore {
} else {
const payload = await this.getMetadataPayload()
const profile = this.getStoredProfile(payload)
created =
!profile || !this.isProfileMatch(profile) ? await this.recreateCollectionForProfile(profile) : false
const compatible = payload?.[KEY.schema] === SCHEMA && profile && this.isProfileMatch(profile)
created = compatible ? false : await this.recreateCollectionForCompatibility(profile)
}
} else {
// Exists but wrong vector size, recreate with enhanced error handling
@@ -474,7 +477,7 @@ export class QdrantVectorStore implements IVectorStore {
if (!payload) {
return false
}
const validKeys = ["filePath", "codeChunk", "startLine", "endLine"]
const validKeys = ["filePath", "fileHash", "codeChunk", "startLine", "endLine"]
const hasValidKeys = validKeys.every((key) => key in payload)
return hasValidKeys
}
@@ -544,7 +547,7 @@ export class QdrantVectorStore implements IVectorStore {
exact: false,
},
with_payload: {
include: ["filePath", "codeChunk", "startLine", "endLine", "pathSegments"],
include: ["filePath", "fileHash", "codeChunk", "startLine", "endLine", "pathSegments"],
},
}
@@ -676,14 +679,7 @@ export class QdrantVectorStore implements IVectorStore {
*/
async hasIndexedData(): Promise<boolean> {
try {
const collectionInfo = await this.getCollectionInfo()
if (!collectionInfo) {
log.info("Qdrant collection has no indexed data", {
collection: this.collectionName,
reason: "collection_missing",
})
return false
}
const collectionInfo = await this.client.getCollection(this.collectionName)
// Check if the collection has any points indexed
const pointsCount = collectionInfo.points_count ?? 0
if (pointsCount === 0) {
@@ -713,8 +709,8 @@ export class QdrantVectorStore implements IVectorStore {
log.info("No indexing metadata marker found. Using backward compatibility mode (checking points_count > 0).")
return pointsCount > 0
} catch (error) {
log.warn("Failed to check if collection has data", { error })
return false
log.error("Failed to check if collection has data", { error })
throw error
}
}
@@ -731,6 +727,7 @@ export class QdrantVectorStore implements IVectorStore {
vector: new Array(this.vectorSize).fill(0),
payload: {
type: "metadata",
[KEY.schema]: SCHEMA,
[KEY.complete]: true,
[KEY.provider]: this.profile.provider,
[KEY.model]: this.profile.modelId,
@@ -761,6 +758,7 @@ export class QdrantVectorStore implements IVectorStore {
vector: new Array(this.vectorSize).fill(0),
payload: {
type: "metadata",
[KEY.schema]: SCHEMA,
[KEY.complete]: false,
[KEY.provider]: this.profile.provider,
[KEY.model]: this.profile.modelId,
packages/kilo-indexing/src/indexing/worktree-overlay.ts
+6 -19
View File
@@ -64,35 +64,22 @@ export class WorktreeOverlay {
this.ready = true
}
async baselineResult(
result: VectorStoreSearchResult,
checks: Map<string, Promise<{ valid: boolean; content: string }>>,
): Promise<boolean> {
async baselineResult(result: VectorStoreSearchResult, checks: Map<string, Promise<boolean>>): Promise<boolean> {
const filePath = result.payload?.filePath
if (typeof filePath !== "string") return false
const rel = this.relative(filePath)
if (!rel || this.shadows.has(rel) || this.blocked.has(rel)) return false
const expected = this.baseline.get(rel)
if (!expected) return false
if (!expected || result.payload?.fileHash !== expected) return false
const existing = checks.get(rel)
const valid =
existing ??
Promise.all([
readFile(path.join(this.baselinePath, ...rel.split("/")), "utf-8"),
readFile(path.join(this.workspacePath, ...rel.split("/")), "utf-8"),
])
.then(([baseline, current]) => {
const hash = (content: string) => createHash("sha256").update(content).digest("hex")
return { valid: hash(baseline) === expected && hash(current) === expected, content: current }
})
.catch(() => ({ valid: false, content: "" }))
readFile(path.join(this.workspacePath, ...rel.split("/")), "utf-8")
.then((content) => createHash("sha256").update(content).digest("hex") === expected)
.catch(() => false)
checks.set(rel, valid)
const checked = await valid
if (!checked.valid) return false
const chunk = result.payload?.codeChunk
return typeof chunk === "string" && checked.content.includes(chunk)
return valid
}
deltaResult(result: VectorStoreSearchResult): boolean {
packages/kilo-indexing/test/kilocode/indexing/manager.test.ts
+55 -1
View File
@@ -1,4 +1,5 @@
import { describe, expect, test } from "bun:test"
import { describe, expect, spyOn, test } from "bun:test"
import { CacheManager } from "../../../src/indexing/cache-manager"
import { CodeIndexManager } from "../../../src/indexing/manager"
import type { IndexingConfigInput } from "../../../src/indexing/config-manager"
import type { IndexingTelemetryEvent, IndexingTelemetryTrigger } from "../../../src/indexing/interfaces/telemetry"
@@ -73,6 +74,59 @@ function createStartError(location = "orchestrator:startIndexing"): IndexingTele
}
describe("CodeIndexManager", () => {
test("falls back when the shared baseline is not ready", async () => {
const mgr = new CodeIndexManager("/tmp/worktree", "/tmp/cache", "/tmp/main")
let closed = 0
const data = mgr as unknown as {
createBaseline(factory: { createVectorStore(): unknown }): Promise<{ store?: unknown }>
}
const baseline = await data.createBaseline({
createVectorStore() {
return {
async openExisting() {
throw new Error("baseline rebuilding")
},
async close() {
closed += 1
},
}
},
})
expect(baseline.store).toBeUndefined()
expect(closed).toBe(1)
})
test("throttles unchanged baseline cache checks between searches", async () => {
const mgr = new CodeIndexManager("/tmp/worktree", "/tmp/cache", "/tmp/main")
const data = createData(mgr) as Data & {
_baselineStamp: string
_baselineSigned: number
_orchestrator: { state: string }
_searchService: { searchIndex(): Promise<[]> }
}
data._baselineStamp = "same"
data._baselineSigned = Date.now()
data._orchestrator = { state: "Indexed" }
data._searchService = {
async searchIndex() {
return []
},
}
const stamp = spyOn(CacheManager.prototype, "stamp").mockResolvedValue("same")
const initialize = spyOn(CacheManager.prototype, "initialize").mockResolvedValue()
try {
await mgr.searchIndex("first")
await mgr.searchIndex("second")
expect(stamp).toHaveBeenCalledTimes(1)
expect(initialize).not.toHaveBeenCalled()
} finally {
stamp.mockRestore()
initialize.mockRestore()
}
})
test("returns standby state before services are initialized", () => {
const mgr = new CodeIndexManager("/tmp/ws", "/tmp/cache")
const data = mgr as unknown as {
packages/kilo-indexing/test/kilocode/indexing/orchestrator.test.ts
+83
View File
@@ -272,6 +272,89 @@ describe("CodeIndexOrchestrator telemetry", () => {
expect(store.closeCount).toBe(1)
})
test("clears stale vectors and hashes before rebuilding an incomplete store", async () => {
const cache = {
clears: 0,
async clearCacheFile() {
this.clears += 1
},
async flush() {},
}
const store = new Store(false, false)
const orchestrator = new CodeIndexOrchestrator(
createConfig(),
new CodeIndexStateManager(),
"/tmp/ws",
cache as unknown as CacheManager,
store as unknown as IVectorStore,
new Scanner(1, 1, 1) as unknown as DirectoryScanner,
new Watcher() as unknown as IFileWatcher,
)
await orchestrator.startIndexing("background")
expect(store.clearCount).toBe(1)
expect(cache.clears).toBe(1)
expect(orchestrator.state).toBe("Indexed")
})
test("does not clear data when index completeness cannot be read", async () => {
const cache = {
clears: 0,
async clearCacheFile() {
this.clears += 1
},
}
const store = new Store(true, false)
store.hasIndexedData = async () => {
throw new Error("metadata unavailable")
}
const orchestrator = new CodeIndexOrchestrator(
createConfig(),
new CodeIndexStateManager(),
"/tmp/ws",
cache as unknown as CacheManager,
store as unknown as IVectorStore,
new Scanner(1, 1, 1) as unknown as DirectoryScanner,
new Watcher() as unknown as IFileWatcher,
)
await orchestrator.startIndexing("background")
expect(store.clearCount).toBe(0)
expect(cache.clears).toBe(0)
expect(orchestrator.state).toBe("Error")
})
test("rebuilds a complete independent index when the shared baseline is unavailable", async () => {
const cache = {
clears: 0,
async clearCacheFile() {
this.clears += 1
},
async flush() {},
}
const store = new Store(true, false)
const orchestrator = new CodeIndexOrchestrator(
createConfig(),
new CodeIndexStateManager(),
"/tmp/ws",
cache as unknown as CacheManager,
store as unknown as IVectorStore,
new Scanner(1, 1, 1) as unknown as DirectoryScanner,
new Watcher() as unknown as IFileWatcher,
undefined,
undefined,
true,
)
await orchestrator.startIndexing("background")
expect(store.clearCount).toBe(1)
expect(cache.clears).toBe(1)
expect(orchestrator.state).toBe("Indexed")
})
test("preserves cache and collection data on retryable start failures", async () => {
const events: IndexingTelemetryEvent[] = []
const cache = {
packages/kilo-indexing/test/kilocode/indexing/search-service.test.ts
+26 -22
View File
@@ -9,10 +9,10 @@ import { CodeIndexSearchService } from "../../../src/indexing/search-service"
import { CodeIndexStateManager } from "../../../src/indexing/state-manager"
import { WorktreeOverlay } from "../../../src/indexing/worktree-overlay"
const result = (filePath: string, score: number, codeChunk = filePath): VectorStoreSearchResult => ({
const result = (filePath: string, score: number, codeChunk = filePath, fileHash?: string): VectorStoreSearchResult => ({
id: `${filePath}:${score}`,
score,
payload: { filePath, codeChunk, startLine: 1, endLine: 1 },
payload: { filePath, fileHash, codeChunk, startLine: 1, endLine: 1 },
})
const store = (results: VectorStoreSearchResult[], limits: number[]): IVectorStore =>
@@ -79,7 +79,7 @@ describe("CodeIndexSearchService worktree search", () => {
embedder(calls),
store([result("src/changed.ts", 0.9, "worktree")], deltaLimits),
{
store: store([result("src/changed.ts", 0.99), result("src/base.ts", 0.8)], baseLimits),
store: store([result("src/changed.ts", 0.99), result("src/base.ts", 0.8, "src/base.ts", hash)], baseLimits),
overlay,
},
)
@@ -105,7 +105,17 @@ describe("CodeIndexSearchService worktree search", () => {
const state = new CodeIndexStateManager()
state.setSystemState("Indexed")
const service = new CodeIndexSearchService(config(), state, embedder([]), store([], []), {
store: store([result("file.ts", 0.99, "export const value = 'primary-only'")], []),
store: store(
[
result(
"file.ts",
0.99,
"export const value = 'primary-only'",
createHash("sha256").update("export const value = 'primary-only'").digest("hex"),
),
],
[],
),
overlay,
})
@@ -144,24 +154,18 @@ describe("CodeIndexSearchService worktree search", () => {
const limits: number[] = []
const state = new CodeIndexStateManager()
state.setSystemState("Indexed")
const service = new CodeIndexSearchService(
config(),
state,
embedder([]),
store([], []),
{
store: store(
[
result("src/a.ts", 0.99),
result("src/b.ts", 0.98),
result("src/c.ts", 0.8),
result("src/d.ts", 0.7),
],
limits,
),
overlay,
},
)
const service = new CodeIndexSearchService(config(), state, embedder([]), store([], []), {
store: store(
[
result("src/a.ts", 0.99),
result("src/b.ts", 0.98),
result("src/c.ts", 0.8, "src/c.ts", hash("src/c.ts")),
result("src/d.ts", 0.7, "src/d.ts", hash("src/d.ts")),
],
limits,
),
overlay,
})
const results = await service.searchIndex("query")
packages/kilo-indexing/test/kilocode/indexing/vector-store/lancedb-vector-store.test.ts
+65 -9
View File
@@ -173,6 +173,29 @@ describe("LocalVectorStore", () => {
expect(mockLoadLanceDB).toHaveBeenCalledTimes(1)
expect(mockLanceDBModule.connect).not.toHaveBeenCalled()
})
test("serializes native connections across stores", async () => {
const other = new LanceDBVectorStore(path.join("mock", "other"), vectorSize, dbDirectory)
store["db"] = null
other["lancedbModule"] = mockLanceDBModule
let active = 0
let maximum = 0
mockLanceDBModule.connect.mockImplementation(async () => {
active += 1
maximum = Math.max(maximum, active)
await Bun.sleep(10)
active -= 1
return mockDb
})
try {
await Promise.all([store.collectionExists(), other.collectionExists()])
} finally {
await other.close()
}
expect(maximum).toBe(1)
})
})
describe("initialize", () => {
@@ -183,7 +206,9 @@ describe("LocalVectorStore", () => {
modelId: "",
dimension: vectorSize,
})
store["_getMetadataValue"] = mock().mockResolvedValue("true")
store["_getMetadataValue"] = mock((_: unknown, key: string) =>
Promise.resolve(key === "index_schema" ? "2" : "true"),
)
await store.openExisting()
@@ -211,19 +236,49 @@ describe("LocalVectorStore", () => {
expect(mockDb.dropTable).toHaveBeenCalled()
})
test("should not recreate if vector size matches", async () => {
test("should not recreate if vector size and schema match", async () => {
mockDb.tableNames.mockResolvedValue(["vector", "metadata"])
mockDb.openTable.mockResolvedValue(mockTable)
store["_getStoredVectorSize"] = mock().mockResolvedValue(vectorSize)
store["_getMetadataValue"] = mock().mockResolvedValue("2")
const result = await store.initialize()
expect(result).toBe(false)
})
test("recreates an index using the legacy payload schema", async () => {
mockDb.tableNames.mockResolvedValue(["vector", "metadata"])
mockDb.openTable.mockResolvedValue(mockTable)
store["_getStoredVectorSize"] = mock().mockResolvedValue(vectorSize)
store["_getMetadataValue"] = mock().mockResolvedValue("1")
expect(await store.initialize()).toBe(true)
expect(mockDb.dropTable).toHaveBeenCalledTimes(2)
})
test("should throw error on LanceDB failure", async () => {
mockDb.tableNames.mockRejectedValue(new Error("fail"))
await expect(store.initialize()).rejects.toThrow()
})
test("does not recreate when vector metadata cannot be read", async () => {
store["_getStoredVectorSize"] = mock().mockRejectedValue(new Error("metadata unavailable"))
await expect(store.initialize()).rejects.toThrow("metadata unavailable")
expect(mockDb.dropTable).not.toHaveBeenCalled()
expect(mockDb.createTable).not.toHaveBeenCalled()
})
test("does not recreate when profile metadata cannot be read", async () => {
mockTable.countRows.mockResolvedValue(1)
store["_getStoredVectorSize"] = mock().mockResolvedValue(vectorSize)
store["_getMetadataValue"] = mock().mockResolvedValue("2")
store["_getStoredEmbeddingProfile"] = mock().mockRejectedValue(new Error("profile unavailable"))
await expect(store.initialize()).rejects.toThrow("profile unavailable")
expect(mockDb.dropTable).not.toHaveBeenCalled()
expect(mockDb.createTable).not.toHaveBeenCalled()
})
test("should recreate tables when stored embedding identity differs", async () => {
const identity = {
provider: "openai",
@@ -332,7 +387,7 @@ describe("LocalVectorStore", () => {
{
id: "123e4567-e89b-12d3-a456-426614174000",
vector: [1, 2, 3],
payload: { filePath: "a", codeChunk: "b", startLine: 1, endLine: 2 },
payload: { filePath: "a", fileHash: "hash-a", codeChunk: "b", startLine: 1, endLine: 2 },
},
]
mockTable.delete.mockResolvedValue(undefined)
@@ -347,7 +402,7 @@ describe("LocalVectorStore", () => {
{
id: "123e4567-e89b-12d3-a456-426614174000",
vector: [1, 2, 3],
payload: { filePath: "a", codeChunk: "b", startLine: 1, endLine: 2 },
payload: { filePath: "a", fileHash: "hash-a", codeChunk: "b", startLine: 1, endLine: 2 },
},
]
mockTable.delete.mockResolvedValue(undefined)
@@ -365,7 +420,7 @@ describe("LocalVectorStore", () => {
distanceRange: distanceRangeSpy,
limit: mock().mockReturnThis(),
toArray: mock().mockResolvedValue([
{ id: "2", _distance: 0.2, filePath: "a", codeChunk: "c", startLine: 3, endLine: 4 },
{ id: "2", _distance: 0.2, filePath: "a", fileHash: "hash-a", codeChunk: "c", startLine: 3, endLine: 4 },
]),
})
const results = await store.search([1, 2, 3], "a", 0.7, 1)
@@ -389,7 +444,7 @@ describe("LocalVectorStore", () => {
distanceRange: distanceRangeSpy,
limit: mock().mockReturnThis(),
toArray: mock().mockResolvedValue([
{ id: "2", _distance: 0.2, filePath: "a", codeChunk: "c", startLine: 3, endLine: 4 },
{ id: "2", _distance: 0.2, filePath: "a", fileHash: "hash-a", codeChunk: "c", startLine: 3, endLine: 4 },
]),
})
const results = await store.search([1, 2, 3], "a", 0.1, 2)
@@ -506,7 +561,7 @@ describe("LocalVectorStore", () => {
})
test("should return true for valid payload", () => {
const payload: Payload = { filePath: "a", codeChunk: "b", startLine: 1, endLine: 2 }
const payload: Payload = { filePath: "a", fileHash: "hash-a", codeChunk: "b", startLine: 1, endLine: 2 }
expect(store["isPayloadValid"](payload)).toBe(true)
})
})
@@ -606,6 +661,7 @@ describe("LocalVectorStore", () => {
vector: [1, 2, 3],
payload: {
filePath: "test.ts",
fileHash: "hash-test",
codeChunk: "code",
startLine: 1,
endLine: 2,
@@ -625,12 +681,12 @@ describe("LocalVectorStore", () => {
{
id: "123e4567-e89b-12d3-a456-426614174000",
vector: [1, 2, 3],
payload: { filePath: "a", codeChunk: "b", startLine: 1, endLine: 2 },
payload: { filePath: "a", fileHash: "hash-a", codeChunk: "b", startLine: 1, endLine: 2 },
},
{
id: "' OR '1'='1",
vector: [4, 5, 6],
payload: { filePath: "c", codeChunk: "d", startLine: 3, endLine: 4 },
payload: { filePath: "c", fileHash: "hash-c", codeChunk: "d", startLine: 3, endLine: 4 },
},
]
packages/kilo-indexing/test/kilocode/indexing/vector-store/qdrant-client.test.ts
+48 -12
View File
@@ -510,6 +510,7 @@ describe("QdrantVectorStore", () => {
mockRetrieve.mockResolvedValue([
{
payload: {
index_schema: 2,
indexing_complete: true,
embedding_provider: "openai",
embedding_model_id: "",
@@ -600,6 +601,33 @@ describe("QdrantVectorStore", () => {
expect(mockCreatePayloadIndex).toHaveBeenCalledTimes(6)
})
test("recreates a populated collection using the legacy payload schema", async () => {
mockGetCollection
.mockResolvedValueOnce({
points_count: 7,
config: { params: { vectors: { size: mockVectorSize } } },
} as any)
.mockRejectedValueOnce({ response: { status: 404 }, message: "Not found" })
mockRetrieve.mockResolvedValue([
{
payload: {
index_schema: 1,
indexing_complete: true,
embedding_provider: "openai",
embedding_model_id: "",
embedding_dimension: mockVectorSize,
},
},
] as any)
mockDeleteCollection.mockResolvedValue(true as any)
mockCreateCollection.mockResolvedValue(true as any)
mockCreatePayloadIndex.mockResolvedValue({} as any)
expect(await vectorStore.initialize()).toBe(true)
expect(mockDeleteCollection).toHaveBeenCalledTimes(1)
expect(mockCreateCollection).toHaveBeenCalledTimes(1)
})
test("should recreate collection when stored embedding identity mismatches", async () => {
const identity = {
provider: "openai",
@@ -1279,6 +1307,7 @@ describe("QdrantVectorStore", () => {
score: 0.85,
payload: {
filePath: "src/test.ts",
fileHash: "test-hash",
codeChunk: "test code",
startLine: 1,
endLine: 5,
@@ -1290,6 +1319,7 @@ describe("QdrantVectorStore", () => {
score: 0.75,
payload: {
filePath: "src/utils.ts",
fileHash: "test-hash",
codeChunk: "utility code",
startLine: 10,
endLine: 15,
@@ -1314,7 +1344,7 @@ describe("QdrantVectorStore", () => {
exact: false,
},
with_payload: {
include: ["filePath", "codeChunk", "startLine", "endLine", "pathSegments"],
include: ["filePath", "fileHash", "codeChunk", "startLine", "endLine", "pathSegments"],
},
})
expect(callArgs.filter).toEqual({
@@ -1334,6 +1364,7 @@ describe("QdrantVectorStore", () => {
score: 0.85,
payload: {
filePath: "src/components/Button.tsx",
fileHash: "test-hash",
codeChunk: "button code",
startLine: 1,
endLine: 5,
@@ -1353,7 +1384,7 @@ describe("QdrantVectorStore", () => {
score_threshold: DEFAULT_SEARCH_MIN_SCORE,
limit: DEFAULT_MAX_SEARCH_RESULTS,
params: { hnsw_ef: 128, exact: false },
with_payload: { include: ["filePath", "codeChunk", "startLine", "endLine", "pathSegments"] },
with_payload: { include: ["filePath", "fileHash", "codeChunk", "startLine", "endLine", "pathSegments"] },
})
expect(callArgs2.filter).toEqual({
must: [
@@ -1385,7 +1416,7 @@ describe("QdrantVectorStore", () => {
exact: false,
},
with_payload: {
include: ["filePath", "codeChunk", "startLine", "endLine", "pathSegments"],
include: ["filePath", "fileHash", "codeChunk", "startLine", "endLine", "pathSegments"],
},
})
expect(callArgs3.filter).toEqual({
@@ -1412,7 +1443,7 @@ describe("QdrantVectorStore", () => {
exact: false,
},
with_payload: {
include: ["filePath", "codeChunk", "startLine", "endLine", "pathSegments"],
include: ["filePath", "fileHash", "codeChunk", "startLine", "endLine", "pathSegments"],
},
})
expect(callArgs4.filter).toEqual({
@@ -1429,6 +1460,7 @@ describe("QdrantVectorStore", () => {
score: 0.85,
payload: {
filePath: "src/test.ts",
fileHash: "test-hash",
codeChunk: "test code",
startLine: 1,
endLine: 5,
@@ -1446,6 +1478,7 @@ describe("QdrantVectorStore", () => {
score: 0.55,
payload: {
filePath: "src/test2.ts",
fileHash: "test-hash",
codeChunk: "test code 2",
startLine: 10,
endLine: 15,
@@ -1472,6 +1505,7 @@ describe("QdrantVectorStore", () => {
score: 0.85,
payload: {
filePath: "src/test.ts",
fileHash: "test-hash",
codeChunk: "test code",
startLine: 1,
endLine: 5,
@@ -1492,6 +1526,7 @@ describe("QdrantVectorStore", () => {
score: 0.55,
payload: {
filePath: "src/test2.ts",
fileHash: "test-hash",
codeChunk: "test code 2",
startLine: 10,
endLine: 15,
@@ -1540,7 +1575,7 @@ describe("QdrantVectorStore", () => {
exact: false,
},
with_payload: {
include: ["filePath", "codeChunk", "startLine", "endLine", "pathSegments"],
include: ["filePath", "fileHash", "codeChunk", "startLine", "endLine", "pathSegments"],
},
})
expect(callArgs5.filter).toEqual({
@@ -1588,6 +1623,7 @@ describe("QdrantVectorStore", () => {
score: 0.85,
payload: {
filePath: "src/test.ts",
fileHash: "test-hash",
codeChunk: "test code",
startLine: 1,
endLine: 5,
@@ -1611,7 +1647,7 @@ describe("QdrantVectorStore", () => {
exact: false,
},
with_payload: {
include: ["filePath", "codeChunk", "startLine", "endLine", "pathSegments"],
include: ["filePath", "fileHash", "codeChunk", "startLine", "endLine", "pathSegments"],
},
})
expect(callArgs7.filter).toEqual({
@@ -1640,7 +1676,7 @@ describe("QdrantVectorStore", () => {
exact: false,
},
with_payload: {
include: ["filePath", "codeChunk", "startLine", "endLine", "pathSegments"],
include: ["filePath", "fileHash", "codeChunk", "startLine", "endLine", "pathSegments"],
},
})
expect(callArgs6.filter).toEqual({
@@ -1667,7 +1703,7 @@ describe("QdrantVectorStore", () => {
exact: false,
},
with_payload: {
include: ["filePath", "codeChunk", "startLine", "endLine", "pathSegments"],
include: ["filePath", "fileHash", "codeChunk", "startLine", "endLine", "pathSegments"],
},
})
expect(callArgs8.filter).toEqual({
@@ -1694,7 +1730,7 @@ describe("QdrantVectorStore", () => {
exact: false,
},
with_payload: {
include: ["filePath", "codeChunk", "startLine", "endLine", "pathSegments"],
include: ["filePath", "fileHash", "codeChunk", "startLine", "endLine", "pathSegments"],
},
})
expect(callArgs9.filter).toEqual({
@@ -1721,7 +1757,7 @@ describe("QdrantVectorStore", () => {
exact: false,
},
with_payload: {
include: ["filePath", "codeChunk", "startLine", "endLine", "pathSegments"],
include: ["filePath", "fileHash", "codeChunk", "startLine", "endLine", "pathSegments"],
},
})
expect(callArgs10.filter).toEqual({
@@ -1748,7 +1784,7 @@ describe("QdrantVectorStore", () => {
exact: false,
},
with_payload: {
include: ["filePath", "codeChunk", "startLine", "endLine", "pathSegments"],
include: ["filePath", "fileHash", "codeChunk", "startLine", "endLine", "pathSegments"],
},
})
expect(callArgs11.filter).toEqual({
@@ -1781,7 +1817,7 @@ describe("QdrantVectorStore", () => {
exact: false,
},
with_payload: {
include: ["filePath", "codeChunk", "startLine", "endLine", "pathSegments"],
include: ["filePath", "fileHash", "codeChunk", "startLine", "endLine", "pathSegments"],
},
})
expect(callArgs12.filter).toEqual({
packages/opencode/src/kilocode/indexing-worker-client.ts
+20 -17
View File
@@ -170,28 +170,31 @@ export namespace IndexingWorker {
)
},
search(query, directoryPrefix) {
return call(
state,
{ type: "request", key, method: "search", input: { query, directoryPrefix } },
(message) => {
if (message.ok && message.method === "search") return message.value
throw new Error("Unexpected indexing worker search response.")
},
)
return call(state, { type: "request", key, method: "search", input: { query, directoryPrefix } }, (message) => {
if (message.ok && message.method === "search") return message.value
throw new Error("Unexpected indexing worker search response.")
})
},
async dispose() {
if (!active || state.stopped) return
active = false
await withTimeout(
call(state, { type: "request", key, method: "dispose", input: undefined }, (message) => {
if (message.ok && message.method === "dispose") return message.value
throw new Error("Unexpected indexing worker dispose response.")
}),
5000,
"Indexing worker reset timed out",
).catch((err) => {
if (state.hosts.get(key) === host) state.hosts.delete(key)
if (pool.get(key) === host) pool.delete(key)
try {
await withTimeout(
call(state, { type: "request", key, method: "dispose", input: undefined }, (message) => {
if (message.ok && message.method === "dispose") return message.value
throw new Error("Unexpected indexing worker dispose response.")
}),
5000,
"Indexing worker reset timed out",
)
} catch (err) {
callbacks.failure(err)
})
} finally {
if (state.hosts.get(key) === host) state.hosts.delete(key)
if (pool.get(key) === host) pool.delete(key)
}
},
}
state.hosts.set(key, host)
packages/opencode/src/kilocode/indexing-worker.ts
+4 -13
View File
@@ -13,7 +13,6 @@ type Entry = {
const managers = new Map<string, Entry>()
const context = new AsyncLocalStorage<string>()
const queues = new Map<string, Promise<void>>()
let initQueue = Promise.resolve()
function send(message: Result | Event) {
postMessage(message)
@@ -75,7 +74,9 @@ async function handle(request: Request) {
}
if (request.method === "search") {
const value = await managers.get(request.key)?.manager.searchIndex(request.input.query, request.input.directoryPrefix)
const value = await managers
.get(request.key)
?.manager.searchIndex(request.input.query, request.input.directoryPrefix)
send({ type: "result", id: request.id, method: "search", ok: true, value: value ?? [] })
return
}
@@ -90,17 +91,7 @@ async function handle(request: Request) {
onmessage = (event: MessageEvent<Request>) => {
const request = event.data
const prior = queues.get(request.key) ?? Promise.resolve()
const task = prior.then(() =>
context.run(request.key, () => {
if (request.method !== "init") return handle(request)
const next = initQueue.then(() => handle(request))
initQueue = next.then(
() => undefined,
() => undefined,
)
return next
}),
)
const task = prior.then(() => context.run(request.key, () => handle(request)))
const queued = task.finally(() => {
if (queues.get(request.key) === queued) queues.delete(request.key)
})
packages/opencode/test/kilocode/indexing-worker.test.ts
+25 -3
View File
@@ -55,7 +55,29 @@ test("routes multiple directories through the shared indexing worker", async ()
expect(failures).toEqual([])
})
test("reuses enabled workers across provider initialization errors", async () => {
test("allows same-directory recreation while disposal is pending", async () => {
await using tmp = await tmpdir()
const hooks = {
status() {},
telemetry() {},
warning() {},
log() {},
failure() {},
}
const first = IndexingWorker.create(tmp.path, tmp.path, hooks)
await first.init({ enabled: false, embedderProvider: "openai" })
const disposing = first.dispose()
const second = IndexingWorker.create(tmp.path, tmp.path, hooks)
const status = await second.init({ enabled: false, embedderProvider: "openai" })
await disposing
await second.dispose()
expect(second).not.toBe(first)
expect(status.state).toBe("Disabled")
})
test("releases enabled workers after provider initialization errors", async () => {
await using tmp = await tmpdir()
const drivers = new Set<IndexingWorker.Driver>()
@@ -107,6 +129,6 @@ test("reuses enabled workers across provider initialization errors", async () =>
await engine.dispose()
expect(status.state).toBe("Disabled")
expect(drivers.has(engine)).toBe(true)
expect(drivers.size).toBe(1)
expect(drivers.has(engine)).toBe(false)
expect(drivers.size).toBe(3)
})