From 42627729b4834efec80d44ad9eabca49584eb659 Mon Sep 17 00:00:00 2001 From: marius-kilocode Date: Tue, 16 Jun 2026 15:56:44 +0200 Subject: [PATCH] fix(cli): refine shared index concurrency --- .../src/indexing/cache-manager.ts | 7 + .../src/indexing/interfaces/vector-store.ts | 1 + .../kilo-indexing/src/indexing/manager.ts | 120 ++++++++++++------ .../src/indexing/orchestrator.ts | 14 +- .../src/indexing/processors/file-watcher.ts | 5 +- .../src/indexing/processors/scanner.ts | 1 + .../src/indexing/search-service.ts | 11 +- .../vector-store/lancedb-vector-store.ts | 108 +++++++++------- .../indexing/vector-store/qdrant-client.ts | 30 ++--- .../src/indexing/worktree-overlay.ts | 25 +--- .../test/kilocode/indexing/manager.test.ts | 56 +++++++- .../kilocode/indexing/orchestrator.test.ts | 83 ++++++++++++ .../kilocode/indexing/search-service.test.ts | 48 +++---- .../vector-store/lancedb-vector-store.test.ts | 74 +++++++++-- .../vector-store/qdrant-client.test.ts | 60 +++++++-- .../src/kilocode/indexing-worker-client.ts | 37 +++--- .../opencode/src/kilocode/indexing-worker.ts | 17 +-- .../test/kilocode/indexing-worker.test.ts | 28 +++- 18 files changed, 513 insertions(+), 212 deletions(-) diff --git a/packages/kilo-indexing/src/indexing/cache-manager.ts b/packages/kilo-indexing/src/indexing/cache-manager.ts index a70d591892..ba5cd3735f 100644 --- a/packages/kilo-indexing/src/indexing/cache-manager.ts +++ b/packages/kilo-indexing/src/indexing/cache-manager.ts @@ -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 { + return fs + .stat(this.cachePath) + .then((value) => `${value.mtimeMs}:${value.ctimeMs}:${value.size}`) + .catch(() => undefined) + } } diff --git a/packages/kilo-indexing/src/indexing/interfaces/vector-store.ts b/packages/kilo-indexing/src/indexing/interfaces/vector-store.ts index 8857ae25db..7712502f14 100644 --- a/packages/kilo-indexing/src/indexing/interfaces/vector-store.ts +++ b/packages/kilo-indexing/src/indexing/interfaces/vector-store.ts @@ -96,6 +96,7 @@ export interface VectorStoreSearchResult { export interface Payload { filePath: string + fileHash?: string codeChunk: string startLine: number endLine: number diff --git a/packages/kilo-indexing/src/indexing/manager.ts b/packages/kilo-indexing/src/indexing/manager.ts index 77b5797787..05e4c5fc15 100644 --- a/packages/kilo-indexing/src/indexing/manager.ts +++ b/packages/kilo-indexing/src/indexing/manager.ts @@ -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 | undefined private _overlay: WorktreeOverlay | undefined private _isRecoveringFromError = false @@ -449,18 +461,36 @@ export class CodeIndexManager { private async refreshBaseline(): Promise { 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 { 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() 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 { + private async _recreateServices(prepared?: Baseline): Promise { 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 }) } diff --git a/packages/kilo-indexing/src/indexing/orchestrator.ts b/packages/kilo-indexing/src/indexing/orchestrator.ts index 46c9b2b069..d7ae84e00f 100644 --- a/packages/kilo-indexing/src/indexing/orchestrator.ts +++ b/packages/kilo-indexing/src/indexing/orchestrator.ts @@ -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, diff --git a/packages/kilo-indexing/src/indexing/processors/file-watcher.ts b/packages/kilo-indexing/src/indexing/processors/file-watcher.ts index 28ae6f457d..eda57909ab 100644 --- a/packages/kilo-indexing/src/indexing/processors/file-watcher.ts +++ b/packages/kilo-indexing/src/indexing/processors/file-watcher.ts @@ -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, diff --git a/packages/kilo-indexing/src/indexing/processors/scanner.ts b/packages/kilo-indexing/src/indexing/processors/scanner.ts index 6f96096fe9..564c0d47c6 100644 --- a/packages/kilo-indexing/src/indexing/processors/scanner.ts +++ b/packages/kilo-indexing/src/indexing/processors/scanner.ts @@ -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, diff --git a/packages/kilo-indexing/src/indexing/search-service.ts b/packages/kilo-indexing/src/indexing/search-service.ts index 71da883498..139c1a41a8 100644 --- a/packages/kilo-indexing/src/indexing/search-service.ts +++ b/packages/kilo-indexing/src/indexing/search-service.ts @@ -58,7 +58,7 @@ export class CodeIndexSearchService { return search(maxResults) })() const base = (async () => { - const checks = new Map>() + const checks = new Map>() const search = async (limit: number): Promise => { 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() 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) { diff --git a/packages/kilo-indexing/src/indexing/vector-store/lancedb-vector-store.ts b/packages/kilo-indexing/src/indexing/vector-store/lancedb-vector-store.ts index c5405d6daf..f97363af68 100644 --- a/packages/kilo-indexing/src/indexing/vector-store/lancedb-vector-store.ts +++ b/packages/kilo-indexing/src/indexing/vector-store/lancedb-vector-store.ts @@ -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(run: () => Promise): Promise { + 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 { - 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 { - 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 { - 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 { - 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 { - 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 { @@ -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 { + 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 { 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 { 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)") diff --git a/packages/kilo-indexing/src/indexing/vector-store/qdrant-client.ts b/packages/kilo-indexing/src/indexing/vector-store/qdrant-client.ts index 7082a0a406..53a4df9156 100644 --- a/packages/kilo-indexing/src/indexing/vector-store/qdrant-client.ts +++ b/packages/kilo-indexing/src/indexing/vector-store/qdrant-client.ts @@ -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 { + private async recreateCollectionForCompatibility(stored?: EmbeddingProfile): Promise { 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 { 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, diff --git a/packages/kilo-indexing/src/indexing/worktree-overlay.ts b/packages/kilo-indexing/src/indexing/worktree-overlay.ts index 28efc0fd29..741bf4ae80 100644 --- a/packages/kilo-indexing/src/indexing/worktree-overlay.ts +++ b/packages/kilo-indexing/src/indexing/worktree-overlay.ts @@ -64,35 +64,22 @@ export class WorktreeOverlay { this.ready = true } - async baselineResult( - result: VectorStoreSearchResult, - checks: Map>, - ): Promise { + async baselineResult(result: VectorStoreSearchResult, checks: Map>): Promise { 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 { diff --git a/packages/kilo-indexing/test/kilocode/indexing/manager.test.ts b/packages/kilo-indexing/test/kilocode/indexing/manager.test.ts index 4999d52df4..9eabad3c61 100644 --- a/packages/kilo-indexing/test/kilocode/indexing/manager.test.ts +++ b/packages/kilo-indexing/test/kilocode/indexing/manager.test.ts @@ -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 { diff --git a/packages/kilo-indexing/test/kilocode/indexing/orchestrator.test.ts b/packages/kilo-indexing/test/kilocode/indexing/orchestrator.test.ts index d2f4369719..4bef1912c7 100644 --- a/packages/kilo-indexing/test/kilocode/indexing/orchestrator.test.ts +++ b/packages/kilo-indexing/test/kilocode/indexing/orchestrator.test.ts @@ -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 = { diff --git a/packages/kilo-indexing/test/kilocode/indexing/search-service.test.ts b/packages/kilo-indexing/test/kilocode/indexing/search-service.test.ts index 1c83734ce0..9447aa688f 100644 --- a/packages/kilo-indexing/test/kilocode/indexing/search-service.test.ts +++ b/packages/kilo-indexing/test/kilocode/indexing/search-service.test.ts @@ -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") diff --git a/packages/kilo-indexing/test/kilocode/indexing/vector-store/lancedb-vector-store.test.ts b/packages/kilo-indexing/test/kilocode/indexing/vector-store/lancedb-vector-store.test.ts index 7355abb830..2ac7fdeeab 100644 --- a/packages/kilo-indexing/test/kilocode/indexing/vector-store/lancedb-vector-store.test.ts +++ b/packages/kilo-indexing/test/kilocode/indexing/vector-store/lancedb-vector-store.test.ts @@ -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 }, }, ] diff --git a/packages/kilo-indexing/test/kilocode/indexing/vector-store/qdrant-client.test.ts b/packages/kilo-indexing/test/kilocode/indexing/vector-store/qdrant-client.test.ts index 36739fb355..9ecb36221b 100644 --- a/packages/kilo-indexing/test/kilocode/indexing/vector-store/qdrant-client.test.ts +++ b/packages/kilo-indexing/test/kilocode/indexing/vector-store/qdrant-client.test.ts @@ -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({ diff --git a/packages/opencode/src/kilocode/indexing-worker-client.ts b/packages/opencode/src/kilocode/indexing-worker-client.ts index 1cc67c36f4..0f055c5db1 100644 --- a/packages/opencode/src/kilocode/indexing-worker-client.ts +++ b/packages/opencode/src/kilocode/indexing-worker-client.ts @@ -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) diff --git a/packages/opencode/src/kilocode/indexing-worker.ts b/packages/opencode/src/kilocode/indexing-worker.ts index bfbca31e99..a37827a631 100644 --- a/packages/opencode/src/kilocode/indexing-worker.ts +++ b/packages/opencode/src/kilocode/indexing-worker.ts @@ -13,7 +13,6 @@ type Entry = { const managers = new Map() const context = new AsyncLocalStorage() const queues = new Map>() -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) => { 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) }) diff --git a/packages/opencode/test/kilocode/indexing-worker.test.ts b/packages/opencode/test/kilocode/indexing-worker.test.ts index 5107495c91..a3c44002ad 100644 --- a/packages/opencode/test/kilocode/indexing-worker.test.ts +++ b/packages/opencode/test/kilocode/indexing-worker.test.ts @@ -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() @@ -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) })