fix(indexing): serialize workspace index writes and preserve partial indexes

.changeset/single-writer-codebase-index.md

@@ -0,0 +1,6 @@
+---
+"@kilocode/kilo-indexing": minor
+---
+
+Keep codebase indexing intact when several sessions share a workspace. Only one session now writes a workspace's index, and a session that finds a partially built index no longer discards it and starts over.
+

packages/kilo-indexing/src/indexing/cache-manager.ts

@@ -19,11 +19,11 @@ export class CacheManager implements ICacheManager {
   private saveTask = Promise.resolve()
 
   constructor(
-    private readonly cacheDirectory: string,
+    public readonly dir: string,
     private readonly workspacePath: string,
   ) {
     const hash = createHash("sha256").update(workspacePath).digest("hex")
-    this.cachePath = path.join(cacheDirectory, `roo-index-cache-${hash}.json`)
+    this.cachePath = path.join(dir, `roo-index-cache-${hash}.json`)
   }
 
   async initialize(): Promise<void> {

packages/kilo-indexing/src/indexing/interfaces/vector-store.ts

@@ -70,10 +70,13 @@ export interface IVectorStore {
   collectionExists(): Promise<boolean>
 
   /**
-   * Checks if the collection exists and has indexed points
-   * @returns Promise resolving to boolean indicating if the collection exists and has points
+   * Reports how far along the store is. `partial` means points exist but no
+   * successful scan finished, which is what a store owned by another live
+   * session looks like. Callers must treat `partial` as usable-but-untrusted
+   * and must never clear it on behalf of a session that does not own the
+   * writer lock.
    */
-  hasIndexedData(): Promise<boolean>
+  state(): Promise<StoreState>
 
   /**
    * Marks the indexing process as complete by storing metadata
@@ -94,6 +97,13 @@ export interface VectorStoreSearchResult {
   payload?: Payload | null
 }
 
+export type StoreStatus = "empty" | "partial" | "complete"
+
+export type StoreState = {
+  status: StoreStatus
+  points?: number
+}
+
 export interface Payload {
   filePath: string
   fileHash?: string

packages/kilo-indexing/src/indexing/manager.ts

@@ -160,6 +160,13 @@ export class CodeIndexManager {
     if (!this.isFeatureEnabled || !this.isFeatureConfigured) return
     if (this._retryTask || this._isRecoveringFromError) return
 
+    // Lock contention is not failure: another live session holds the writer
+    // lock, so retrying with backoff only spins while the holder watches.
+    // The second session retries on its own triggers (settings, restart,
+    // watcher events); a retry here would also recreate services, tearing
+    // down the loser's search path for no gain.
+    if (event.error.includes("Another session is indexing this workspace")) return
+
     if (this._retryAttempt >= this._retryMaxAttempts) {
       log.warn("indexing recovery retries exhausted", {
         workspacePath: this.workspacePath,

packages/kilo-indexing/src/indexing/orchestrator.ts

@@ -11,21 +11,30 @@ import type {
   IndexingTelemetryTrigger,
 } from "./interfaces/telemetry"
 import type { IVectorStore } from "./interfaces/vector-store"
+import type { StoreState } from "./interfaces"
 import { DirectoryScanner } from "./processors"
 import type { CacheManager } from "./cache-manager"
 import type { Disposable } from "./runtime"
 import { Log } from "../util/log"
 import { sanitizeErrorMessage } from "./shared/validation-helpers"
 import { DEFAULT_VECTOR_STORE } from "./constants"
 import type { WorktreeOverlay } from "./worktree-overlay"
+import { acquire, type Writer } from "./writer-lock"
 
 const log = Log.create({ service: "indexing-orchestrator" })
 
+const EMPTY: StoreState = { status: "empty" }
+
 export class CodeIndexOrchestrator {
   private _fileWatcherSubscriptions: Disposable[] = []
   private _isProcessing = false
   private _cancelRequested = false
   private _active?: Promise<void>
+  // The session's writer lock, held from the first successful run until
+  // shutdown (or release on error/clear). The file watcher keeps writing the
+  // store after every scan, so releasing per-run would let a second session
+  // acquire the lock and scan concurrently — the corruption the lock targets.
+  private _writer?: Writer
 
   constructor(
     private readonly configManager: CodeIndexConfigManager,
@@ -97,6 +106,11 @@ export class CodeIndexOrchestrator {
             workspacePath: this.workspacePath,
             filesInBatch: paths.length,
           })
+          // The watcher writes the store on its own after the scan, outside
+          // runIndexing. Re-check ownership before every batch: a heartbeat can
+          // lapse (event loop stall) long enough for another session to reclaim
+          // the lock, and this session must not keep committing without it.
+          void this._verifyWriter()
           if (this.stateManager.state !== "Indexing") {
             this.stateManager.setSystemState("Indexing", "Processing file changes...")
           }
@@ -145,6 +159,68 @@ export class CodeIndexOrchestrator {
     return task
   }
 
+  /**
+   * Only one session may write a workspace's index. Without this, a second
+   * session starts a scan against a store the first one is still filling,
+   * clears it as unusable, and leaves both sessions committing to the same
+   * table. The lock is session-held (see _writer): once acquired, later runs
+   * reuse it instead of contending with themselves.
+   */
+  private async _acquire(): Promise<Writer | undefined> {
+    if (this._writer) {
+      // The session caches its writer across runs. Verify ownership before
+      // reusing it: a heartbeat that saw another session's token stops beating,
+      // but only a fresh check here stops this orchestrator from writing the
+      // store after it has lost the lock.
+      if (await this._writer.owned()) return this._writer
+      log.warn("lost the writer lock since the last run; reacquiring", {
+        workspacePath: this.workspacePath,
+      })
+      this._writer = undefined
+    }
+    const dir = this.cacheManager.dir
+    if (!dir) {
+      log.warn("no cache directory for the writer lock, proceeding unlocked", {
+        workspacePath: this.workspacePath,
+      })
+      return { release: async () => {}, owned: async () => true }
+    }
+    const writer = await acquire(dir, this.workspacePath)
+    if (writer) {
+      this._writer = writer
+      return writer
+    }
+    log.info("writer lock is held elsewhere", { workspacePath: this.workspacePath })
+    return undefined
+  }
+
+  private async _releaseWriter(): Promise<void> {
+    const writer = this._writer
+    this._writer = undefined
+    await writer?.release()
+  }
+
+  /**
+   * Stops the watcher when this session no longer owns the writer lock.
+   *
+   * The heartbeat stops beating after a takeover, but the watcher keeps
+   * receiving file events and writing the store. Checking once per batch
+   * bounds how much can be written after ownership is lost; shutting the
+   * watcher down turns a silent split-brain into a visible Error state.
+   */
+  private async _verifyWriter(): Promise<void> {
+    const writer = this._writer
+    if (!writer) return
+    if (await writer.owned()) return
+    if (this._writer !== writer) return
+    this._writer = undefined
+    log.error("writer lock lost; stopping the file watcher to avoid unlocked writes", {
+      workspacePath: this.workspacePath,
+    })
+    this.stateManager.setSystemState("Error", "Lost the indexing writer lock to another session. File watching stopped.")
+    this.stopWatcher()
+  }
+
   private async runIndexing(trigger: IndexingTelemetryTrigger): Promise<void> {
     log.info("indexing start requested", {
       workspacePath: this.workspacePath,
@@ -176,6 +252,7 @@ export class CodeIndexOrchestrator {
     }
 
     this._cancelRequested = false
+
     this._isProcessing = true
     this.stateManager.setSystemState("Indexing", "Initializing services...")
 
@@ -184,6 +261,27 @@ export class CodeIndexOrchestrator {
     let mode: IndexingTelemetryMode | undefined
 
     try {
+      // Acquired inside the try and session-held (see _writer): a throw between
+      // acquire and the body (e.g. from state setup below) releases below and
+      // must not leave the lock fresh for STALE_MS, blocking other sessions.
+      // The lock is NOT released on success: the file watcher keeps writing the
+      // store after the scan, so it stays held until shutdown or clearIndexData.
+      const writer = await this._acquire()
+      if (!writer) {
+        // Contention is worth observing, not a silent park: the manager
+        // surfaces start events as activity, so emit an error event the manager
+        // can retry or display rather than leaving the session in Standby with
+        // no record of the missed run.
+        this.emitError(
+          "orchestrator:startIndexing",
+          new Error("Another session is indexing this workspace."),
+          "scan",
+          trigger,
+        )
+        this.stateManager.setSystemState("Standby", "Another session is indexing this workspace.")
+        return
+      }
+
       this.overlay?.prepare()
       await this._startWatcher()
 
@@ -214,8 +312,11 @@ export class CodeIndexOrchestrator {
         })
       }
 
-      const hasExistingData = this.overlay ? false : await this.vectorStore.hasIndexedData()
-      if (!this.overlay && !hasExistingData) {
+      const index = this.overlay ? EMPTY : await this.vectorStore.state()
+      // Only an empty store is safe to clear. A partial one belongs to a run
+      // that died or is still going, and clearing it throws away every point
+      // it managed to embed.
+      if (!this.overlay && index.status === "empty") {
         if (!collectionCreated) await this.vectorStore.clearCollection()
         await this.cacheManager.clearCacheFile()
         log.info("cleared indexing cache before full scan", {
@@ -225,7 +326,7 @@ export class CodeIndexOrchestrator {
       }
       log.info("checked vector store indexed data", {
         workspacePath: this.workspacePath,
-        hasExistingData,
+        status: index.status,
         collectionCreated,
       })
 
@@ -234,7 +335,7 @@ export class CodeIndexOrchestrator {
         return
       }
 
-      mode = hasExistingData && !collectionCreated ? "incremental" : "full"
+      mode = index.status === "complete" && !collectionCreated ? "incremental" : "full"
 
       if (mode === "incremental") {
         log.info("collection has existing data, running incremental scan")
@@ -245,7 +346,7 @@ export class CodeIndexOrchestrator {
       } else {
         log.info("running full scan", {
           workspacePath: this.workspacePath,
-          hasExistingData,
+          status: index.status,
           collectionCreated,
         })
         this.stateManager.setSystemState("Indexing", "Services ready. Starting workspace scan...")
@@ -265,6 +366,9 @@ export class CodeIndexOrchestrator {
       const msg = err instanceof Error ? err.message : "Unknown error"
       this.stateManager.setSystemState("Error", `Failed during initial scan: ${msg}`)
       this.stopWatcher()
+      // The watcher is stopped, so this session no longer writes the store:
+      // hand the lock back instead of holding it while wedged in Error.
+      await this._releaseWriter()
     } finally {
       this._isProcessing = false
       log.info("indexing start flow finished", {
@@ -359,7 +463,10 @@ export class CodeIndexOrchestrator {
     this.overlay?.reconcile(this.cacheManager.getAllHashes())
     await this.cacheManager.flush?.()
     await this.vectorStore.markIndexingComplete()
-    await this.vectorStore.close?.()
+    // The scan finished, but this session's file watcher keeps writing the
+    // store for every file change, so the session-held writer lock stays
+    // acquired until shutdown or clearIndexData (see _writer). Releasing here
+    // would let a second session acquire it and scan concurrently.
     this.fileWatcher.setCollecting(true)
     this.stateManager.setSystemState("Indexed", "File watcher started. Index up-to-date.")
     log.info("workspace scan finalized", {
@@ -385,13 +492,23 @@ export class CodeIndexOrchestrator {
   public async shutdown(): Promise<void> {
     this._cancelRequested = true
     this.scanner.cancel()
+    // Stop the session's own watcher writes before releasing the writer lock:
+    // the lock releases below, so the store must be quiet by then rather than
+    // still receiving file-watcher batches.
+    if (this.fileWatcher.shutdown) await this.fileWatcher.shutdown()
+    else this.fileWatcher.dispose()
     this.fileWatcher.setCollecting(false)
+    // Wait out the in-flight scan before closing the store: a scan that is
+    // still running would lazily reconnect on its next getDb()/getTable() call,
+    // reopening a connection close() just tore down and leaking it. Closing
+    // after _active settles keeps the guarantee without racing the scan. The
+    // next session cannot interleave here: it can only acquire the writer lock
+    // once _releaseWriter runs below.
     await this._active
+    await this.vectorStore.close?.()
+    await this._releaseWriter()
     for (const sub of this._fileWatcherSubscriptions) sub.dispose()
     this._fileWatcherSubscriptions = []
-    if (this.fileWatcher.shutdown) await this.fileWatcher.shutdown()
-    else this.fileWatcher.dispose()
-    await this.vectorStore.close?.()
     this._isProcessing = false
   }
 
@@ -413,17 +530,67 @@ export class CodeIndexOrchestrator {
     log.info("cancelling indexing", { workspacePath: this.workspacePath })
     this._cancelRequested = true
     this.scanner.cancel()
-    this.stopWatcher()
+    // Keep the file watcher alive: cancel only aborts the scan, and the
+    // session still holds the writer lock, so watcher batches continue to own
+    // the store. Disposing here (via stopWatcher) would strand _active's
+    // finally with a disposed watcher while file changes keep arriving.
     this.stateManager.setSystemState("Standby", "Indexing cancelled.")
     this._isProcessing = false
     log.info("indexing cancelled", { workspacePath: this.workspacePath })
   }
 
   public async clearIndexData(): Promise<void> {
+    // Reuse the session-held lock when this session owns it: acquiring again
+    // would contend with ourselves. Whatever the source, the watcher is
+    // stopped below, so no writes remain and the lock is released afterwards.
+    const writer = await this._acquire()
+    if (!writer) {
+      // A user-requested clear must not vanish silently when another session
+      // holds the lock: surface contention as an error event with a
+      // clear-specific location. The manager only auto-retries startIndexing
+      // locations, so this stays terminal: the user sees the failure and can
+      // retry the clear once the other session finishes.
+      this.emitError(
+        "orchestrator:clearIndexData",
+        new Error("Another session is indexing this workspace; index data was not cleared."),
+        "scan",
+      )
+      this.stateManager.setSystemState("Error", "Another session is indexing this workspace; index data was not cleared.")
+      return
+    }
+
     this._isProcessing = true
     log.info("clearing index data", { workspacePath: this.workspacePath })
 
     try {
+      // A clear must not race an in-flight scan. Cancel it and wait for the
+      // scan to settle before deleting the collection: otherwise batches still
+      // running can re-add points after the clear and write under a lock this
+      // method is about to release (shutdown() waits the same way).
+      this._cancelRequested = true
+      this.scanner.cancel()
+      await this._active
+      // Verify ownership before deleting even when a writer is still cached:
+      // a heartbeat lapse during the awaited scan can let another session
+      // reclaim the lock while _writer stays set. _acquire() re-checks owned()
+      // and reuses the cached writer only when this session still holds it.
+      const retry = await this._acquire()
+      if (!retry) {
+        this.emitError(
+          "orchestrator:clearIndexData",
+          new Error("Another session is indexing this workspace; index data was not cleared."),
+          "scan",
+        )
+        this.stateManager.setSystemState(
+          "Error",
+          "Another session is indexing this workspace; index data was not cleared.",
+        )
+        return
+      }
+      // The watcher's drainTask is separate from _active: shutdown() waits for
+      // it before returning, so a batch already writing cannot upsert after
+      // deleteCollection() or after the lock is released below.
+      if (this.fileWatcher.shutdown) await this.fileWatcher.shutdown()
       this.stopWatcher()
 
       try {
@@ -444,6 +611,9 @@ export class CodeIndexOrchestrator {
       }
     } finally {
       this._isProcessing = false
+      // stopWatcher() above means this session no longer writes the store, so
+      // a session-held lock must be handed back rather than kept until exit.
+      await this._releaseWriter()
       log.info("finished clearing index data", {
         workspacePath: this.workspacePath,
         state: this.stateManager.state,

packages/kilo-indexing/src/indexing/processors/file-watcher.ts

@@ -317,8 +317,10 @@ export class FileWatcher implements IFileWatcher {
    */
   async shutdown(): Promise<void> {
     this.collecting = false
-    if (this.batchProcessDebounceTimer) clearTimeout(this.batchProcessDebounceTimer)
-    this.batchProcessDebounceTimer = undefined
+    if (this.batchProcessDebounceTimer) {
+      clearTimeout(this.batchProcessDebounceTimer)
+      this.batchProcessDebounceTimer = undefined
+    }
     const subscription = this.subscription
     this.subscription = undefined
     await subscription