fix(personal): serialize add/remove/reload on an async job lock

The #5558 fix took the job lock INSIDE the threadpool worker and only on the
add path, so (1) remove_directory and /reload mutated PersonalDocsManager's
unsynchronized list/index concurrently with an in-flight add — the inconsistent
state the PR claimed to prevent — and (2) a queued add blocked on the lock while
holding an AnyIO threadpool token, starving the shared pool.

Move the lock to an asyncio.Lock acquired in the async handler BEFORE offloading,
and route add, remove and reload through it. A waiting request now parks on the
event loop instead of pinning a worker, and all three mutators are serialized so
the 'add/remove are serialized and cannot leave inconsistent state' guarantee
holds. remove and reload also run their blocking work off the event loop. The
lock is per-router so each app binds it to its own loop; single-process scope.

Tests: add-vs-remove and add-vs-reload serialization regressions (async via
ASGITransport, since asyncio.Lock deadlocks starlette TestClient's portal); the
existing add-vs-add test converted to the same driver.

routes/personal_routes.py

@@ -1,9 +1,9 @@
 # routes/personal_routes.py
 """Routes for personal documents management."""
+import asyncio
 import os
 import logging
 import shutil
-import threading
 import uuid
 from typing import Any, Dict, List, Tuple
 from fastapi import APIRouter, HTTPException, Query, Request, UploadFile, File, Depends
@@ -20,14 +20,6 @@ UPLOADS_DIR = PERSONAL_UPLOADS_DIR
 
 logger = logging.getLogger(__name__)
 
-# Serializes directory index jobs across requests. Now that indexing runs in
-# the threadpool (#5558), concurrent requests would otherwise run in parallel
-# and race PersonalDocsManager's unsynchronized list mutations and file
-# writes; before the threadpool move they serialized on the blocked event
-# loop, so one-at-a-time is behavior parity.
-_index_job_lock = threading.Lock()
-
-
 def _personal_upload_dir_for_owner(owner: str | None, *, create: bool = True) -> str:
     """Return the per-owner upload directory used for direct RAG uploads."""
     owner_segment = secure_filename((owner or "local").strip())[:80] or "local"
@@ -150,6 +142,22 @@ def setup_personal_routes(personal_docs_manager, rag_manager, rag_available):
     """
     router = APIRouter(prefix="/api/personal")
 
+    # Serializes directory index jobs across requests. Indexing runs in the
+    # threadpool (#5558), so concurrent requests would otherwise run in parallel
+    # and race PersonalDocsManager's unsynchronized list mutations and file
+    # writes; before the threadpool move they serialized on the blocked event
+    # loop, so one-at-a-time is behavior parity.
+    #
+    # An asyncio.Lock acquired in the async handler BEFORE offloading: a waiting
+    # request parks on the event loop instead of pinning a threadpool worker (an
+    # earlier threading.Lock taken INSIDE the worker meant queued jobs held pool
+    # tokens while blocked, starving every other run_in_threadpool caller).
+    # add/remove/reload all take this lock, so their mutations never interleave.
+    # Per-router (not module-global) so each app binds it to its own event loop.
+    # Scope is the single process: multi-worker deployments would need a shared
+    # lock (out of scope for #5558).
+    _index_job_lock = asyncio.Lock()
+
     def _rag():
         """Get the current RAG manager, retrying init if needed."""
         return get_rag_manager()
@@ -181,8 +189,12 @@ def setup_personal_routes(personal_docs_manager, rag_manager, rag_available):
         return {"files": files, "directories": directories}
     
     @router.post("/reload")
-    def api_personal_reload(owner: str = Depends(require_user), _admin: None = Depends(require_admin)):
-        personal_docs_manager.refresh_index()
+    async def api_personal_reload(owner: str = Depends(require_user), _admin: None = Depends(require_admin)):
+        # refresh_index() re-extracts text across every tracked directory —
+        # blocking work. Take the shared job lock (so it cannot race an add /
+        # remove) and run it off the event loop.
+        async with _index_job_lock:
+            await run_in_threadpool(personal_docs_manager.refresh_index)
         return {"ok": True, "count": len(personal_docs_manager.index)}
     
     @router.post("/add_directory")
@@ -217,21 +229,23 @@ def setup_personal_routes(personal_docs_manager, rag_manager, rag_available):
             rag = _rag()
             if rag:
                 def _index_directory():
-                    with _index_job_lock:
-                        result = rag.index_personal_documents(directory, owner=owner)
-                        if result["success"]:
-                            # Also update the personal_docs_manager to track
-                            # this directory. Kept inside the threadpool call:
-                            # it triggers refresh_index(), which re-extracts
-                            # text across tracked directories.
-                            personal_docs_manager.add_directory(directory, index=False)
-                        return result
+                    result = rag.index_personal_documents(directory, owner=owner)
+                    if result["success"]:
+                        # Also update the personal_docs_manager to track this
+                        # directory. Kept inside the offloaded call: it triggers
+                        # refresh_index(), which re-extracts text across tracked
+                        # directories.
+                        personal_docs_manager.add_directory(directory, index=False)
+                    return result
 
                 # Indexing walks, embeds, and stores the whole tree — minutes
                 # on a real directory. The handler is async, so calling it
                 # inline runs it on the event loop and every other request
-                # queues behind it until it finishes (#5558).
-                result = await run_in_threadpool(_index_directory)
+                # queues behind it until it finishes (#5558). Serialize on the
+                # async job lock BEFORE offloading so a queued request parks on
+                # the loop instead of pinning a threadpool worker.
+                async with _index_job_lock:
+                    result = await run_in_threadpool(_index_directory)
 
                 if result["success"]:
                     return {
@@ -272,17 +286,25 @@ def setup_personal_routes(personal_docs_manager, rag_manager, rag_available):
 
             logger.info(f"Removing directory from RAG: {directory}")
 
-            # Always remove from personal_docs_manager tracking
-            if hasattr(personal_docs_manager, 'remove_directory'):
-                personal_docs_manager.remove_directory(directory)
-
-            # Remove from RAG vector store (best-effort)
             rag = _rag()
-            if rag:
-                try:
-                    rag.remove_directory(directory)
-                except Exception as e:
-                    logger.warning(f"RAG removal failed for directory {directory}: {e}")
+
+            def _remove_directory():
+                # Always remove from personal_docs_manager tracking. This
+                # mutates the same unsynchronized list/index an add job touches
+                # and re-extracts text (refresh_index), so it is blocking work.
+                if hasattr(personal_docs_manager, 'remove_directory'):
+                    personal_docs_manager.remove_directory(directory)
+                # Remove from RAG vector store (best-effort).
+                if rag:
+                    try:
+                        rag.remove_directory(directory)
+                    except Exception as e:
+                        logger.warning(f"RAG removal failed for directory {directory}: {e}")
+
+            # Same job lock as add/reload so remove cannot interleave with an
+            # in-flight add; offloaded off the event loop.
+            async with _index_job_lock:
+                await run_in_threadpool(_remove_directory)
 
             return {
                 "success": True,

tests/test_add_directory_event_loop.py

@@ -11,14 +11,43 @@ triggers ``refresh_index()``, which re-extracts text across tracked dirs.
 These tests build the real router with fake managers and compare the thread
 the indexing work runs on against the event loop's thread.
 """
+import asyncio
 import os
 import threading
 
 os.environ.setdefault("DATABASE_URL", "sqlite:///:memory:")
 
+import httpx
 from fastapi import FastAPI
 from fastapi.testclient import TestClient
 
+
+def _serialization_probe():
+    """Shared counter proving two critical sections never overlap."""
+    state = {"active": 0, "max_active": 0}
+    lock = threading.Lock()
+
+    def enter():
+        with lock:
+            state["active"] += 1
+            state["max_active"] = max(state["max_active"], state["active"])
+
+    def leave():
+        with lock:
+            state["active"] -= 1
+
+    return state, enter, leave
+
+
+# Concurrency tests are `async def` (pyproject asyncio_mode="auto") and drive the
+# ASGI app through httpx.ASGITransport + AsyncClient + asyncio.gather, NOT starlette
+# TestClient + ThreadPoolExecutor: the job lock is an asyncio.Lock acquired in the
+# async handler, and TestClient's portal-thread dispatch deadlocks against it (same
+# reason test_notes_fail_closed_auth.py uses ASGITransport). asyncio.gather runs both
+# requests on the test's own loop.
+def _async_client(app):
+    return httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://t")
+
 import routes.personal_routes as personal_routes
 from core.middleware import require_admin
 from src.auth_helpers import require_user
@@ -104,43 +133,30 @@ def test_response_and_bookkeeping_unchanged(tmp_path, monkeypatch):
     assert record["bookkeeping_index_flag"] is False
 
 
-def test_concurrent_add_directory_requests_serialize_indexing(tmp_path, monkeypatch):
+async def test_concurrent_add_directory_requests_serialize_indexing(tmp_path, monkeypatch):
     """Off-loop execution must not mean parallel index jobs: concurrent
     requests would race PersonalDocsManager's unsynchronized list mutations
     and file writes (save_directories/_save_excluded are plain open('w'))."""
     import time
-    from concurrent.futures import ThreadPoolExecutor
 
-    record = {}
-    state = {"active": 0, "max_active": 0}
-    state_lock = threading.Lock()
+    state, enter, leave = _serialization_probe()
 
     def _slow_index(self, directory, owner=None):
-        with state_lock:
-            state["active"] += 1
-            state["max_active"] = max(state["max_active"], state["active"])
-        time.sleep(0.2)
-        with state_lock:
-            state["active"] -= 1
+        enter(); time.sleep(0.2); leave()
         return {"success": True, "indexed_count": 1, "failed_count": 0}
 
     monkeypatch.setattr(_FakeRag, "index_personal_documents", _slow_index)
 
+    record = {}
     app = _build_app(tmp_path, monkeypatch, record)
     for name in ("docs_a", "docs_b"):
         (tmp_path / name).mkdir()
 
-    client = TestClient(app)
-    with ThreadPoolExecutor(max_workers=2) as pool:
-        futures = [
-            pool.submit(
-                client.post,
-                "/api/personal/add_directory",
-                json={"directory": str(tmp_path / name)},
-            )
-            for name in ("docs_a", "docs_b")
-        ]
-        results = [f.result() for f in futures]
+    async with _async_client(app) as ac:
+        results = await asyncio.gather(
+            ac.post("/api/personal/add_directory", json={"directory": str(tmp_path / "docs_a")}),
+            ac.post("/api/personal/add_directory", json={"directory": str(tmp_path / "docs_b")}),
+        )
 
     assert all(r.status_code == 200 for r in results)
     assert state["max_active"] == 1, (
@@ -164,3 +180,74 @@ def test_failed_indexing_still_returns_500(tmp_path, monkeypatch):
     resp = client.post("/api/personal/add_directory", json={"directory": str(target)})
     assert resp.status_code == 500
     assert "boom" in resp.json()["detail"]
+
+
+async def test_add_and_remove_serialize(tmp_path, monkeypatch):
+    """#5634: remove must hold the SAME job lock as add. Otherwise a remove
+    running while an add job is in flight races PersonalDocsManager's
+    unsynchronized list/index mutations — the inconsistent state the PR's
+    'add/remove are serialized' guarantee claims to prevent."""
+    import time
+
+    state, enter, leave = _serialization_probe()
+
+    def _slow_index(self, directory, owner=None):
+        enter(); time.sleep(0.25); leave()
+        return {"success": True, "indexed_count": 1, "failed_count": 0}
+
+    def _slow_remove(self, directory):
+        enter(); time.sleep(0.25); leave()
+
+    monkeypatch.setattr(_FakeRag, "index_personal_documents", _slow_index)
+    monkeypatch.setattr(_FakeDocsManager, "remove_directory", _slow_remove, raising=False)
+
+    record = {}
+    app = _build_app(tmp_path, monkeypatch, record)
+    (tmp_path / "docs_a").mkdir()
+    (tmp_path / "docs_b").mkdir()
+
+    async with _async_client(app) as ac:
+        results = await asyncio.gather(
+            ac.post("/api/personal/add_directory", json={"directory": str(tmp_path / "docs_a")}),
+            ac.delete("/api/personal/remove_directory", params={"directory": str(tmp_path / "docs_b")}),
+        )
+
+    assert all(r.status_code == 200 for r in results)
+    assert state["max_active"] == 1, (
+        f"{state['max_active']} add/remove critical sections overlapped — "
+        "remove must hold the same index job lock as add"
+    )
+
+
+async def test_reload_serializes_with_add(tmp_path, monkeypatch):
+    """#5634: POST /reload rebuilds the index via refresh_index(); it must hold
+    the same job lock so it cannot race an in-flight add job."""
+    import time
+
+    state, enter, leave = _serialization_probe()
+
+    def _slow_index(self, directory, owner=None):
+        enter(); time.sleep(0.25); leave()
+        return {"success": True, "indexed_count": 1, "failed_count": 0}
+
+    def _slow_refresh(self):
+        enter(); time.sleep(0.25); leave()
+
+    monkeypatch.setattr(_FakeRag, "index_personal_documents", _slow_index)
+    monkeypatch.setattr(_FakeDocsManager, "refresh_index", _slow_refresh, raising=False)
+
+    record = {}
+    app = _build_app(tmp_path, monkeypatch, record)
+    (tmp_path / "docs_a").mkdir()
+
+    async with _async_client(app) as ac:
+        results = await asyncio.gather(
+            ac.post("/api/personal/add_directory", json={"directory": str(tmp_path / "docs_a")}),
+            ac.post("/api/personal/reload"),
+        )
+
+    assert all(r.status_code == 200 for r in results)
+    assert state["max_active"] == 1, (
+        f"{state['max_active']} add/reload critical sections overlapped — "
+        "reload must hold the same index job lock as add"
+    )