fix: atomic session slots, partial-index cleanup, watcher serialization

- server.py: the MAX_ACTIVE_BOTS check ran in offer() while the increment
  happened inside the detached bot task, several awaits later, so concurrent
  offers all passed the guard together. Capacity is now reserved atomically
  under a lock before the offer is handled; ownership transfers to the bot task
  once it is created, and offer() releases the slot on every path where it was
  not. run_interview_bot wraps its whole body so even the early readiness raise
  releases. Measured with 10 simultaneous offers: peak active went 10 -> 2,
  with 8 x 503 and no leaked slots.

- indexer.ts: a rebuild that threw after the stale delete left the documents it
  had already upserted in the index, while the caller dropped the persisted
  cache — the only record of their ids — stranding them in search results.
  Extracted discardPartialIndex(), now used by the cancel path and the catch.
  If that cleanup itself fails it clears discardedPreviousIndex so the cache is
  kept and a later rebuild can retry, rather than losing the ids entirely.

- indexer.ts: the `indexing` guard only stopped watcher work from starting, so
  an upsert/remove already awaiting a file read could write during a rebuild's
  cleanup. Added a generation counter bumped per rebuild and re-checked after
  every await before any write, plus in-flight op tracking that rebuild drains
  before it mutates anything.

- .env.example: MAX_ACTIVE_BOTS now precedes SESSION_HANDSHAKE_TIMEOUT_SECS.

Verified: 10-way offer burst and the reserve-then-fail path; rebuild throwing
mid-scan with cleanup succeeding and failing; and a watcher write racing a
rebuild with and without the generation guard.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: samanyugoyal2010

apps/moss-interview-coach/backend/.env.example

@@ -21,11 +21,11 @@ PIPER_VOICE=en_US-lessac-medium
 
 # Grader subprocess
 GRADE_SUBPROCESS_TIMEOUT_SECS=60
+# Concurrent interviews. Each loads its own STT/TTS and shares one Ollama.
+MAX_ACTIVE_BOTS=2
 # Ends a session whose client never finishes the WebRTC/RTVI handshake, so a
 # dropped offer cannot hold Whisper/Piper/Ollama open.
 SESSION_HANDSHAKE_TIMEOUT_SECS=45
-# Concurrent interviews. Each loads its own STT/TTS and shares one Ollama.
-MAX_ACTIVE_BOTS=2
 
 # Backend
 # Loopback by default: /api/offer is unauthenticated, and each call spins up

apps/moss-interview-coach/backend/server.py

@@ -109,6 +109,27 @@ def build_system_prompt(track_id: str) -> str:
 moss_indexes_ready: dict[str, bool] = {tid: False for tid in INTERVIEW_TRACKS}
 moss_ready = False
 active_bots = 0
+# Guards active_bots. The capacity check and the increment have to be one
+# critical section: offer() runs concurrently, and the bot task that used to do
+# the incrementing is spawned several awaits later, so checking there let any
+# number of simultaneous offers pass the limit together.
+active_bots_lock = asyncio.Lock()
+
+
+async def reserve_bot_slot() -> bool:
+    """Take a slot if one is free. Caller must release exactly once."""
+    global active_bots
+    async with active_bots_lock:
+        if active_bots >= MAX_ACTIVE_BOTS:
+            return False
+        active_bots += 1
+        return True
+
+
+async def release_bot_slot() -> None:
+    global active_bots
+    async with active_bots_lock:
+        active_bots = max(0, active_bots - 1)
 
 
 class ActiveSession:
@@ -744,7 +765,18 @@ async def run_interview_bot(
     webrtc_connection: SmallWebRTCConnection,
     track_id: str = DEFAULT_TRACK_ID,
 ) -> None:
-    global active_bots
+    # The slot is reserved by offer() before this task is created, so every exit
+    # path here — including the readiness check below — must release it.
+    try:
+        await _run_interview_bot(webrtc_connection, track_id)
+    finally:
+        await release_bot_slot()
+
+
+async def _run_interview_bot(
+    webrtc_connection: SmallWebRTCConnection,
+    track_id: str = DEFAULT_TRACK_ID,
+) -> None:
     track_id = normalize_track_id(track_id)
     if moss_client is None or not moss_indexes_ready.get(track_id):
         raise RuntimeError(
@@ -756,7 +788,6 @@ async def run_interview_bot(
     index_name = track["index_name"]
     system_prompt = build_system_prompt(track_id)
 
-    active_bots += 1
     session: ActiveSession | None = None
     try:
         transport = SmallWebRTCTransport(
@@ -931,7 +962,6 @@ async def _handshake_watchdog(current: ActiveSession) -> None:
     finally:
         if session is not None:
             active_sessions.discard(session)
-        active_bots = max(0, active_bots - 1)
 
 
 async def ensure_moss_loaded() -> None:
@@ -1127,53 +1157,65 @@ async def offer(request: Request) -> dict[str, Any]:
             status_code=503,
             detail=f"Grader worker missing at {GRADER_WORKER_PATH.name}.",
         )
-    # Refuse before spawning: every session loads Whisper/Piper and competes for
+    # Reserve before spawning: every session loads Whisper/Piper and competes for
     # the same local Ollama, so unbounded offers would degrade the live ones.
-    if active_bots >= MAX_ACTIVE_BOTS:
+    # Taken here rather than inside the bot task — that runs several awaits
+    # later, so concurrent offers would all pass a mere check and overshoot.
+    if not await reserve_bot_slot():
         raise HTTPException(
             status_code=503,
             detail=(
-                f"At capacity: {active_bots} interview(s) already running "
+                f"At capacity: {MAX_ACTIVE_BOTS} interview(s) already running "
                 f"(MAX_ACTIVE_BOTS={MAX_ACTIVE_BOTS}). Try again shortly."
             ),
         )
-
-    body = await _json_object_body(request)
-
-    async def webrtc_connection_callback(connection: SmallWebRTCConnection) -> None:
-        # Detached deliberately, not a Starlette background task. Those are
-        # awaited as part of the request lifecycle, and uvicorn waits for
-        # outstanding request tasks *before* running lifespan shutdown — with
-        # timeout_graceful_shutdown defaulting to None, that wait is unbounded.
-        # An interview attached to the request would therefore hang Ctrl-C
-        # forever and the lifespan cleanup below would never get to cancel it.
-        task = asyncio.create_task(
-            run_interview_bot(connection, track_id),
-            name=f"moss-interview-{track_id}",
-        )
-        bot_tasks.add(task)
-        task.add_done_callback(_on_bot_task_done)
-
+    # Ownership of the reserved slot moves to the bot task once it is created;
+    # until then this request must hand it back on every failure path.
+    slot_handed_over = False
     try:
-        answer = await small_webrtc_handler.handle_web_request(
-            request=SmallWebRTCRequest.from_dict(body),
-            webrtc_connection_callback=webrtc_connection_callback,
-        )
-    except HTTPException:
-        # Already carries an intended status; do not flatten it to a 500.
-        raise
-    except (KeyError, TypeError, ValueError) as exc:
-        # A malformed offer is the caller's mistake, not a server fault.
-        logger.warning(f"Malformed WebRTC offer: {exc}")
-        raise HTTPException(
-            status_code=422, detail=f"Malformed WebRTC offer: {exc}"
-        ) from exc
-    except Exception as exc:  # noqa: BLE001
-        logger.exception("Failed to handle WebRTC offer")
-        raise HTTPException(
-            status_code=500,
-            detail="Failed to handle WebRTC offer",
-        ) from exc
+        body = await _json_object_body(request)
+
+        async def webrtc_connection_callback(connection: SmallWebRTCConnection) -> None:
+            # Detached deliberately, not a Starlette background task. Those are
+            # awaited as part of the request lifecycle, and uvicorn waits for
+            # outstanding request tasks *before* running lifespan shutdown — with
+            # timeout_graceful_shutdown defaulting to None, that wait is unbounded.
+            # An interview attached to the request would therefore hang Ctrl-C
+            # forever and the lifespan cleanup below would never get to cancel it.
+            nonlocal slot_handed_over
+            task = asyncio.create_task(
+                run_interview_bot(connection, track_id),
+                name=f"moss-interview-{track_id}",
+            )
+            # From here run_interview_bot's finally releases the slot, so this
+            # request must not — even if handle_web_request fails afterwards.
+            slot_handed_over = True
+            bot_tasks.add(task)
+            task.add_done_callback(_on_bot_task_done)
+
+        try:
+            answer = await small_webrtc_handler.handle_web_request(
+                request=SmallWebRTCRequest.from_dict(body),
+                webrtc_connection_callback=webrtc_connection_callback,
+            )
+        except HTTPException:
+            # Already carries an intended status; do not flatten it to a 500.
+            raise
+        except (KeyError, TypeError, ValueError) as exc:
+            # A malformed offer is the caller's mistake, not a server fault.
+            logger.warning(f"Malformed WebRTC offer: {exc}")
+            raise HTTPException(
+                status_code=422, detail=f"Malformed WebRTC offer: {exc}"
+            ) from exc
+        except Exception as exc:  # noqa: BLE001
+            logger.exception("Failed to handle WebRTC offer")
+            raise HTTPException(
+                status_code=500,
+                detail="Failed to handle WebRTC offer",
+            ) from exc
+    finally:
+        if not slot_handed_over:
+            await release_bot_slot()
 
     return answer
 

apps/moss-vscode/src/indexer/indexer.ts

@@ -26,6 +26,10 @@ export class CodebaseIndexer {
   private watchingEnabled = false;
   /** True once rebuild() has begun destroying the previous index. */
   private discardedPreviousIndex = false;
+  /** Bumped by each rebuild so watcher work started earlier can bail out. */
+  private generation = 0;
+  /** Watcher operations already past their `indexing` guard. */
+  private watcherOps = new Set<Promise<void>>();
   private onPersist: (() => void) | undefined;
 
   setPersistHandler(handler: (() => void) | undefined): void {
@@ -110,8 +114,12 @@ export class CodebaseIndexer {
     }
     this.indexing = true;
     this.discardedPreviousIndex = false;
+    // Invalidate watcher work already in flight, then wait for it to settle so
+    // no late write lands between our delete and our rewrite.
+    this.generation += 1;
 
     try {
+      await this.drainWatcherOps();
       const files = await scanWorkspaceFiles(token);
       this.setStatus({ state: "indexing", processed: 0, total: files.length });
 
@@ -216,16 +224,7 @@ export class CodebaseIndexer {
         // snapshot is already gone (stale docs were deleted above), so discard
         // the partial index rather than presenting it as ready. The user re-runs
         // indexing from the "unindexed" state.
-        const partialIds: string[] = [];
-        for (const [rel, count] of this.pathChunkCounts) {
-          for (let i = 0; i < count; i++) {
-            partialIds.push(`${rel}#chunk-${i}`);
-          }
-        }
-        if (partialIds.length) {
-          await this.deleteInBatches(partialIds);
-        }
-        this.pathChunkCounts.clear();
+        await this.discardPartialIndex();
         this.watchingEnabled = false;
         this.setStatus({ state: "unindexed" });
         return;
@@ -247,6 +246,14 @@ export class CodebaseIndexer {
       });
     } catch (err) {
       const message = err instanceof Error ? err.message : String(err);
+      // If the throw landed after the stale delete, documents from this run may
+      // already be in the index while the persisted cache — soon to be dropped
+      // by the caller — is the only record of their ids. Remove them now so
+      // nothing is stranded answering searches for files that no longer exist.
+      if (this.discardedPreviousIndex) {
+        await this.discardPartialIndex();
+      }
+      this.watchingEnabled = false;
       this.setStatus({ state: "error", message });
       throw err;
     } finally {
@@ -263,17 +270,55 @@ export class CodebaseIndexer {
     }
   }
 
+  /**
+   * Register a watcher operation so a rebuild can wait for it to finish.
+   *
+   * The `indexing` guard only stops watcher work from *starting*; an operation
+   * already awaiting a file read would otherwise run its writes concurrently
+   * with a rebuild's cleanup.
+   */
+  private trackWatcherOp(run: () => Promise<void>): Promise<void> {
+    const op = run();
+    this.watcherOps.add(op);
+    void op.catch(() => undefined).finally(() => this.watcherOps.delete(op));
+    return op;
+  }
+
+  /** Let watcher work that predates this rebuild settle before mutating. */
+  private async drainWatcherOps(): Promise<void> {
+    while (this.watcherOps.size) {
+      const inflight = Array.from(this.watcherOps);
+      await Promise.allSettled(inflight);
+      for (const op of inflight) {
+        this.watcherOps.delete(op);
+      }
+    }
+  }
+
   async upsertFile(uri: vscode.Uri): Promise<void> {
     if (!this.session || !this.watchingEnabled || this.indexing) {
       return;
     }
+    return this.trackWatcherOp(() => this.applyUpsert(uri, this.session!));
+  }
+
+  private async applyUpsert(uri: vscode.Uri, session: LocalMossSession): Promise<void> {
+    // A rebuild starting mid-operation invalidates everything below: its
+    // pathChunkCounts are being rewritten and its documents deleted, so writing
+    // here would resurrect ids the rebuild has already accounted for.
+    const generation = this.generation;
+    const stale = () => generation !== this.generation;
+
     const relativePath = toWorkspaceRelative(uri);
     if (isExcludedFromIndex(relativePath)) {
       return;
     }
     const file = await readFileForIndex(uri);
+    if (stale()) {
+      return;
+    }
     if (!file) {
-      await this.removeFile(uri);
+      await this.applyRemove(uri, session);
       return;
     }
 
@@ -287,14 +332,25 @@ export class CodebaseIndexer {
         toDelete.push(`${file.relativePath}#chunk-${i}`);
       }
       if (toDelete.length) {
-        await this.session.deleteDocs(toDelete);
+        await session.deleteDocs(toDelete);
+        if (stale()) {
+          return;
+        }
       }
     }
 
     if (chunks.length) {
-      await this.session.addDocs(chunks, { upsert: true });
+      await session.addDocs(chunks, { upsert: true });
+      // Re-checked after the write: if a rebuild took over we must not record
+      // these ids, and it will delete them as part of its own cleanup.
+      if (stale()) {
+        return;
+      }
       this.pathChunkCounts.set(file.relativePath, next);
     } else {
+      if (stale()) {
+        return;
+      }
       this.pathChunkCounts.delete(file.relativePath);
     }
 
@@ -306,16 +362,26 @@ export class CodebaseIndexer {
     if (!this.session || !this.watchingEnabled || this.indexing) {
       return;
     }
+    return this.trackWatcherOp(() => this.applyRemove(uri, this.session!));
+  }
+
+  private async applyRemove(uri: vscode.Uri, session: LocalMossSession): Promise<void> {
+    const generation = this.generation;
+    const stale = () => generation !== this.generation;
+
     const relativePath = toWorkspaceRelative(uri);
     const count = this.pathChunkCounts.get(relativePath) ?? 0;
     if (!count) {
       // Best-effort: try deleting a reasonable number of chunks
       const guessIds = Array.from({ length: 64 }, (_, i) => `${relativePath}#chunk-${i}`);
-      await this.session.deleteDocs(guessIds).catch(() => undefined);
+      await session.deleteDocs(guessIds).catch(() => undefined);
       return;
     }
     const ids = Array.from({ length: count }, (_, i) => `${relativePath}#chunk-${i}`);
-    await this.session.deleteDocs(ids);
+    await session.deleteDocs(ids);
+    if (stale()) {
+      return;
+    }
     this.pathChunkCounts.delete(relativePath);
     this.refreshReadyStatus();
     this.requestPersist();
@@ -397,6 +463,32 @@ export class CodebaseIndexer {
     });
   }
 
+  /**
+   * Remove every document this rebuild added and forget their ids.
+   *
+   * If the delete fails, `discardedPreviousIndex` is cleared so the caller
+   * keeps the persisted cache: the ids stay recorded there for a later rebuild
+   * to retry, rather than leaving documents in the index that nothing knows
+   * how to remove.
+   */
+  private async discardPartialIndex(): Promise<void> {
+    const partialIds: string[] = [];
+    for (const [rel, count] of this.pathChunkCounts) {
+      for (let i = 0; i < count; i++) {
+        partialIds.push(`${rel}#chunk-${i}`);
+      }
+    }
+    if (partialIds.length) {
+      try {
+        await this.deleteInBatches(partialIds);
+      } catch {
+        this.discardedPreviousIndex = false;
+        return;
+      }
+    }
+    this.pathChunkCounts.clear();
+  }
+
   private async deleteInBatches(ids: string[]): Promise<void> {
     if (!this.session) {
       return;