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.

.env.example

@@ -189,7 +189,6 @@ SEARXNG_INSTANCE=http://localhost:8080
 # ODYSSEUS_EMAIL_COMPOSE_UPLOAD_MAX_BYTES=26214400       # email compose attachment (25 MB)
 # ODYSSEUS_STT_MAX_AUDIO_BYTES=26214400                  # speech-to-text audio (25 MB)
 # ODYSSEUS_ICS_MAX_BYTES=10485760                        # calendar .ics import (10 MB)
-# ODYSSEUS_TTS_CACHE_MAX_BYTES=524288000                 # TTS cache (500 MB)
 
 # ============================================================
 # Host Docker access (explicit opt-in)

.github/scripts/check-issue-description.js

@@ -153,16 +153,6 @@ module.exports = async ({ github, context, core }) => {
     }
   }
 
-  const LABEL_BAD  = 'needs more info';
-  const LABEL_GOOD = 'ready for review';
-
-  // Closed issues are no longer awaiting review.
-  // This also prevents later edits to closed issues from restoring the label.
-  if (issue.state === 'closed') {
-    await dropLabel(LABEL_GOOD);
-    return;
-  }
-
   // ── Find existing bot comment to update in-place ──────────────────────────
   const MARKER = '<!-- issue-description-check -->';
   const { data: comments } = await github.rest.issues.listComments({
@@ -170,6 +160,9 @@ module.exports = async ({ github, context, core }) => {
   });
   const existing = comments.find(c => c.user.type === 'Bot' && c.body.includes(MARKER));
 
+  const LABEL_BAD  = 'needs more info';
+  const LABEL_GOOD = 'ready for review';
+
   if (failures.length === 0) {
     if (existing) {
       await github.rest.issues.deleteComment({ owner, repo, comment_id: existing.id });

.github/workflows/issue-description-check.yml

@@ -2,7 +2,7 @@ name: ci / issue description check
 
 on:
   issues:
-    types: [opened, edited, reopened, closed]
+    types: [opened, edited, reopened]
 
 permissions:
   issues: write

app.py

@@ -692,7 +692,7 @@ from routes.history.history_routes import setup_history_routes
 app.include_router(setup_history_routes(session_manager, upload_handler=upload_handler))
 
 # Search
-from routes.search.search_routes import setup_search_routes
+from routes.search_routes import setup_search_routes
 app.include_router(setup_search_routes(config))
 
 # Presets

docker-compose.gpu-amd.yml

@@ -67,7 +67,6 @@ services:
       - ODYSSEUS_EMAIL_COMPOSE_UPLOAD_MAX_BYTES=${ODYSSEUS_EMAIL_COMPOSE_UPLOAD_MAX_BYTES:-26214400}
       - ODYSSEUS_STT_MAX_AUDIO_BYTES=${ODYSSEUS_STT_MAX_AUDIO_BYTES:-26214400}
       - ODYSSEUS_ICS_MAX_BYTES=${ODYSSEUS_ICS_MAX_BYTES:-10485760}
-      - ODYSSEUS_TTS_CACHE_MAX_BYTES=${ODYSSEUS_TTS_CACHE_MAX_BYTES}
       - DATA_BRAVE_API_KEY=${DATA_BRAVE_API_KEY:-}
       - GOOGLE_API_KEY=${GOOGLE_API_KEY:-}
       - GOOGLE_PSE_CX=${GOOGLE_PSE_CX:-}

docker-compose.gpu-nvidia.yml

@@ -66,7 +66,6 @@ services:
       - ODYSSEUS_EMAIL_COMPOSE_UPLOAD_MAX_BYTES=${ODYSSEUS_EMAIL_COMPOSE_UPLOAD_MAX_BYTES:-26214400}
       - ODYSSEUS_STT_MAX_AUDIO_BYTES=${ODYSSEUS_STT_MAX_AUDIO_BYTES:-26214400}
       - ODYSSEUS_ICS_MAX_BYTES=${ODYSSEUS_ICS_MAX_BYTES:-10485760}
-      - ODYSSEUS_TTS_CACHE_MAX_BYTES=${ODYSSEUS_TTS_CACHE_MAX_BYTES}
       - DATA_BRAVE_API_KEY=${DATA_BRAVE_API_KEY:-}
       - GOOGLE_API_KEY=${GOOGLE_API_KEY:-}
       - GOOGLE_PSE_CX=${GOOGLE_PSE_CX:-}

docker-compose.yml

@@ -55,7 +55,6 @@ services:
       - ODYSSEUS_EMAIL_COMPOSE_UPLOAD_MAX_BYTES=${ODYSSEUS_EMAIL_COMPOSE_UPLOAD_MAX_BYTES:-26214400}
       - ODYSSEUS_STT_MAX_AUDIO_BYTES=${ODYSSEUS_STT_MAX_AUDIO_BYTES:-26214400}
       - ODYSSEUS_ICS_MAX_BYTES=${ODYSSEUS_ICS_MAX_BYTES:-10485760}
-      - ODYSSEUS_TTS_CACHE_MAX_BYTES=${ODYSSEUS_TTS_CACHE_MAX_BYTES}
       - DATA_BRAVE_API_KEY=${DATA_BRAVE_API_KEY:-}
       - GOOGLE_API_KEY=${GOOGLE_API_KEY:-}
       - GOOGLE_PSE_CX=${GOOGLE_PSE_CX:-}

requirements.txt

@@ -38,10 +38,7 @@ python-dateutil
 caldav
 cryptography
 bcrypt
-# Built-in servers use the v1 low-level Server decorator API. MCP SDK v2 is a
-# breaking rewrite, so keep fresh installs on the maintained v1 line until the
-# servers are migrated together.
-mcp<2
+mcp
 pyotp
 qrcode[pil]
 croniter

routes/personal_routes.py

@@ -1,11 +1,13 @@
 # routes/personal_routes.py
 """Routes for personal documents management."""
+import asyncio
 import os
 import logging
 import shutil
 import uuid
 from typing import Any, Dict, List, Tuple
 from fastapi import APIRouter, HTTPException, Query, Request, UploadFile, File, Depends
+from fastapi.concurrency import run_in_threadpool
 from src.request_models import DirectoryRequest
 from core.constants import BASE_DIR, PERSONAL_DIR, PERSONAL_UPLOADS_DIR
 from src.rag_singleton import get_rag_manager
@@ -18,7 +20,6 @@ UPLOADS_DIR = PERSONAL_UPLOADS_DIR
 
 logger = logging.getLogger(__name__)
 
-
 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"
@@ -141,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()
@@ -172,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")
@@ -207,12 +228,26 @@ def setup_personal_routes(personal_docs_manager, rag_manager, rag_available):
             # Use the RAGManager to index the directory
             rag = _rag()
             if rag:
-                result = rag.index_personal_documents(directory, owner=owner)
-                
+                def _index_directory():
+                    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). 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"]:
-                    # Also update the personal_docs_manager to track this directory
-                    personal_docs_manager.add_directory(directory, index=False)
-                    
                     return {
                         "success": True,
                         "message": f"Successfully indexed {result['indexed_count']} chunks from {directory}",
@@ -251,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,

routes/search/__init__.py

@@ -1,5 +0,0 @@
-"""Search route domain package (slice 2j, #4082/#4071).
-
-Contains search_routes.py, migrated from the flat routes/ directory.
-Backward-compat shim at routes/search_routes.py re-exports from here.
-"""

routes/search/search_routes.py

@@ -1,111 +0,0 @@
-"""Search routes — /api/search/config GET, /api/search POST."""
-
-import logging
-from typing import Dict, Any
-
-from fastapi import APIRouter, Request
-
-import time
-
-from services.search import get_search_config, comprehensive_web_search, PROVIDER_INFO
-from services.search.core import _call_provider
-from services.search.providers import _get_provider_key, _get_search_instance
-
-logger = logging.getLogger(__name__)
-
-
-async def _request_values(request: Request) -> Dict[str, Any]:
-    """Accept JSON, form data, or query params for search endpoints.
-
-    The browser UI posts FormData, while the agent's generic app_api tool
-    posts JSON. FastAPI Form(...) rejects JSON with a 422 before our handler
-    runs, which made the model think SearXNG was broken.
-    """
-    values: Dict[str, Any] = dict(request.query_params)
-    content_type = (request.headers.get("content-type") or "").lower()
-    try:
-        if "application/json" in content_type:
-            body = await request.json()
-            if isinstance(body, dict):
-                values.update(body)
-        else:
-            form = await request.form()
-            values.update(dict(form))
-    except Exception:
-        pass
-    return values
-
-
-def setup_search_routes(config) -> APIRouter:
-    router = APIRouter(tags=["search"])
-
-    @router.get("/api/search/config")
-    async def get_search_settings() -> Dict[str, Any]:
-        return get_search_config()
-
-    @router.post("/api/search")
-    async def do_web_search(request: Request) -> Dict[str, Any]:
-        """Standalone web search — returns context string + source list.
-
-        Used by Compare mode to pre-search once and share results across panes.
-        """
-        values = await _request_values(request)
-        query = str(values.get("query") or values.get("q") or "").strip()
-        if not query:
-            return {"context": "", "sources": [], "error": "query is required"}
-        time_filter = values.get("time_filter") or values.get("freshness")
-        if time_filter is not None:
-            time_filter = str(time_filter).strip() or None
-        try:
-            context, sources = comprehensive_web_search(
-                query, return_sources=True, time_filter=time_filter,
-            )
-            return {"context": context, "sources": sources}
-        except Exception as e:
-            logger.error(f"Standalone web search failed: {e}")
-            return {"context": "", "sources": [], "error": str(e)}
-
-    @router.get("/api/search/providers")
-    async def list_search_providers():
-        """Return available search providers with config status."""
-        providers = []
-        for pid, (label, needs_key, needs_url) in PROVIDER_INFO.items():
-            if pid == "disabled":
-                continue
-            available = True
-            if needs_key and not _get_provider_key(pid):
-                available = False
-            if needs_url and pid == "searxng" and not _get_search_instance():
-                available = False
-            providers.append({
-                "id": pid,
-                "label": label,
-                "available": available,
-            })
-        return providers
-
-    @router.post("/api/search/query")
-    async def search_with_provider(request: Request) -> Dict[str, Any]:
-        """Search using a specific provider. Used by compare search mode."""
-        values = await _request_values(request)
-        query = str(values.get("query") or values.get("q") or "").strip()
-        provider = str(values.get("provider") or "").strip()
-        try:
-            count = int(values.get("count") or values.get("limit") or 10)
-        except Exception:
-            count = 10
-        if not query:
-            return {"results": [], "provider": provider, "error": "query is required"}
-        if provider not in PROVIDER_INFO or provider == "disabled":
-            return {"results": [], "provider": provider, "error": "Unknown provider"}
-        t0 = time.time()
-        try:
-            results = _call_provider(provider, query, min(count, 20))
-            elapsed = round(time.time() - t0, 2)
-            return {"results": results, "provider": provider, "time": elapsed}
-        except Exception as e:
-            elapsed = round(time.time() - t0, 2)
-            logger.error(f"Search provider {provider} failed: {e}")
-            return {"results": [], "provider": provider, "time": elapsed, "error": str(e)}
-
-    return router

routes/search_routes.py

@@ -1,13 +1,111 @@
-"""Backward-compat shim — canonical location is routes/search/search_routes.py.
+"""Search routes — /api/search/config GET, /api/search POST."""
 
-This module is replaced in ``sys.modules`` by the canonical module object so
-that ``import routes.search_routes`` and ``from routes.search_routes import X``
-keep resolving to the canonical module. Keeps existing import paths working
-after slice 2j (#4082/#4071).
-"""
+import logging
+from typing import Dict, Any
 
-import sys as _sys
+from fastapi import APIRouter, Request
 
-from routes.search import search_routes as _canonical  # noqa: F401
+import time
 
-_sys.modules[__name__] = _canonical
+from services.search import get_search_config, comprehensive_web_search, PROVIDER_INFO
+from services.search.core import _call_provider
+from services.search.providers import _get_provider_key, _get_search_instance
+
+logger = logging.getLogger(__name__)
+
+
+async def _request_values(request: Request) -> Dict[str, Any]:
+    """Accept JSON, form data, or query params for search endpoints.
+
+    The browser UI posts FormData, while the agent's generic app_api tool
+    posts JSON. FastAPI Form(...) rejects JSON with a 422 before our handler
+    runs, which made the model think SearXNG was broken.
+    """
+    values: Dict[str, Any] = dict(request.query_params)
+    content_type = (request.headers.get("content-type") or "").lower()
+    try:
+        if "application/json" in content_type:
+            body = await request.json()
+            if isinstance(body, dict):
+                values.update(body)
+        else:
+            form = await request.form()
+            values.update(dict(form))
+    except Exception:
+        pass
+    return values
+
+
+def setup_search_routes(config) -> APIRouter:
+    router = APIRouter(tags=["search"])
+
+    @router.get("/api/search/config")
+    async def get_search_settings() -> Dict[str, Any]:
+        return get_search_config()
+
+    @router.post("/api/search")
+    async def do_web_search(request: Request) -> Dict[str, Any]:
+        """Standalone web search — returns context string + source list.
+
+        Used by Compare mode to pre-search once and share results across panes.
+        """
+        values = await _request_values(request)
+        query = str(values.get("query") or values.get("q") or "").strip()
+        if not query:
+            return {"context": "", "sources": [], "error": "query is required"}
+        time_filter = values.get("time_filter") or values.get("freshness")
+        if time_filter is not None:
+            time_filter = str(time_filter).strip() or None
+        try:
+            context, sources = comprehensive_web_search(
+                query, return_sources=True, time_filter=time_filter,
+            )
+            return {"context": context, "sources": sources}
+        except Exception as e:
+            logger.error(f"Standalone web search failed: {e}")
+            return {"context": "", "sources": [], "error": str(e)}
+
+    @router.get("/api/search/providers")
+    async def list_search_providers():
+        """Return available search providers with config status."""
+        providers = []
+        for pid, (label, needs_key, needs_url) in PROVIDER_INFO.items():
+            if pid == "disabled":
+                continue
+            available = True
+            if needs_key and not _get_provider_key(pid):
+                available = False
+            if needs_url and pid == "searxng" and not _get_search_instance():
+                available = False
+            providers.append({
+                "id": pid,
+                "label": label,
+                "available": available,
+            })
+        return providers
+
+    @router.post("/api/search/query")
+    async def search_with_provider(request: Request) -> Dict[str, Any]:
+        """Search using a specific provider. Used by compare search mode."""
+        values = await _request_values(request)
+        query = str(values.get("query") or values.get("q") or "").strip()
+        provider = str(values.get("provider") or "").strip()
+        try:
+            count = int(values.get("count") or values.get("limit") or 10)
+        except Exception:
+            count = 10
+        if not query:
+            return {"results": [], "provider": provider, "error": "query is required"}
+        if provider not in PROVIDER_INFO or provider == "disabled":
+            return {"results": [], "provider": provider, "error": "Unknown provider"}
+        t0 = time.time()
+        try:
+            results = _call_provider(provider, query, min(count, 20))
+            elapsed = round(time.time() - t0, 2)
+            return {"results": results, "provider": provider, "time": elapsed}
+        except Exception as e:
+            elapsed = round(time.time() - t0, 2)
+            logger.error(f"Search provider {provider} failed: {e}")
+            return {"results": [], "provider": provider, "time": elapsed, "error": str(e)}
+
+    return router

services/tts/tts_service.py

@@ -2,7 +2,6 @@
 """Multi-provider TTS service — dispatches to local Kokoro, OpenAI-compatible API, or browser."""
 
 import io
-import os
 import wave
 import logging
 import hashlib
@@ -42,11 +41,6 @@ class TTSService:
         self.cache_dir = Path(cache_dir)
         self.cache_dir.mkdir(parents=True, exist_ok=True)
         self._kokoro = None  # lazy-init
-        
-        try:
-            self.max_cache_bytes = int(os.getenv("ODYSSEUS_TTS_CACHE_MAX_BYTES", 500 * 1024 * 1024))
-        except ValueError:
-            self.max_cache_bytes = 500 * 1024 * 1024
 
     # ── Settings ──
 
@@ -95,53 +89,6 @@ class TTSService:
         ext = ".mp3" if (len(data) >= 3 and (data[:3] == b'ID3' or (data[0] == 0xff and (data[1] & 0xe0) == 0xe0))) else ".wav"
         (self.cache_dir / f"{key}{ext}").write_bytes(data)
 
-        self._enforce_cache_limit()
-
-    def _enforce_cache_limit(self):
-            """Evicts oldest files if the cache exceeds the configured byte limit."""
-            if self.max_cache_bytes <= 0:
-                return
-
-            try:
-                files = []
-                total_size = 0
-
-                # Safely scan files and sum sizes, ignoring files deleted mid-scan
-                for f in self.cache_dir.iterdir():
-                    try:
-                        if f.is_file() and f.suffix.lower() in (".mp3", ".wav"):
-                            files.append(f)
-                            total_size += f.stat().st_size
-                    except OSError:
-                        continue
-
-                if total_size > self.max_cache_bytes:
-                    logger.info(
-                        f"TTS cache ({total_size} bytes) exceeded limit ({self.max_cache_bytes} bytes). Evicting oldest files."
-                    )
-
-                    # Sort files by modification time (oldest first)
-                    try:
-                        files.sort(key=lambda f: f.stat().st_mtime)
-                    except OSError as e:
-                        logger.warning(f"Failed to sort cache files by mtime: {e}")
-
-                    # Trim down to 80% of max capacity
-                    target_size = self.max_cache_bytes * 0.8
-
-                    while files and total_size > target_size:
-                        f = files.pop(0)
-                        try:
-                            size = f.stat().st_size
-                            f.unlink()
-                            total_size -= size
-                        except OSError as e:
-                            logger.warning(f"Failed to evict cache file {f}: {e}")
-                            continue
-
-            except Exception as e:
-                logger.warning(f"Error enforcing TTS cache limit: {e}", exc_info=True)
-
     def clear_cache(self):
         count = 0
         for f in self.cache_dir.glob("*.*"):

tests/test_add_directory_event_loop.py

@@ -0,0 +1,253 @@
+"""Regression guard for #5558 — POST /api/personal/add_directory must not run
+the indexing job on the event loop.
+
+The handler is ``async def`` but called ``rag.index_personal_documents``
+(os.walk + file reads + per-chunk embedding + Chroma inserts) inline, so
+FastAPI ran the whole job on the event loop and every other request queued
+behind it: indexing a real directory froze the UI and API for 25+ minutes.
+``personal_docs_manager.add_directory`` sits in the same blocking section — it
+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
+
+
+class _FakeRag:
+    def __init__(self, record):
+        self._record = record
+
+    def index_personal_documents(self, directory, owner=None):
+        self._record["index_thread"] = threading.get_ident()
+        return {"success": True, "indexed_count": 3, "failed_count": 0}
+
+
+class _FakeDocsManager:
+    def __init__(self, record):
+        self._record = record
+        self.index = []
+
+    def add_directory(self, directory, *, index=True, owner=None):
+        self._record["bookkeeping_thread"] = threading.get_ident()
+        self._record["bookkeeping_index_flag"] = index
+
+
+def _build_app(tmp_path, monkeypatch, record):
+    monkeypatch.setattr(personal_routes, "PERSONAL_DIR", str(tmp_path))
+    monkeypatch.setattr(personal_routes, "get_rag_manager", lambda: _FakeRag(record))
+
+    app = FastAPI()
+    app.include_router(
+        personal_routes.setup_personal_routes(_FakeDocsManager(record), None, True)
+    )
+    app.dependency_overrides[require_user] = lambda: "tester"
+    app.dependency_overrides[require_admin] = lambda: None
+
+    @app.get("/loop-thread")
+    async def loop_thread_probe():
+        return {"thread": threading.get_ident()}
+
+    return app
+
+
+def test_indexing_runs_off_the_event_loop(tmp_path, monkeypatch):
+    record = {}
+    app = _build_app(tmp_path, monkeypatch, record)
+    target = tmp_path / "docs"
+    target.mkdir()
+
+    # Context-manager client: one portal/event loop serves both requests, so
+    # the probe and the POST are guaranteed to see the same loop thread.
+    with TestClient(app) as client:
+        loop_thread = client.get("/loop-thread").json()["thread"]
+        resp = client.post(
+            "/api/personal/add_directory", json={"directory": str(target)}
+        )
+
+    assert resp.status_code == 200
+    assert record["index_thread"] != loop_thread, (
+        "index_personal_documents ran on the event loop thread — every other "
+        "request queues behind the indexing job (#5558)"
+    )
+    assert record["bookkeeping_thread"] != loop_thread, (
+        "personal_docs_manager.add_directory (refresh_index) ran on the event "
+        "loop thread"
+    )
+
+
+def test_response_and_bookkeeping_unchanged(tmp_path, monkeypatch):
+    record = {}
+    app = _build_app(tmp_path, monkeypatch, record)
+    target = tmp_path / "docs"
+    target.mkdir()
+
+    client = TestClient(app)
+    resp = client.post("/api/personal/add_directory", json={"directory": str(target)})
+
+    assert resp.status_code == 200
+    body = resp.json()
+    assert body["success"] is True
+    assert body["indexed_count"] == 3
+    assert body["failed_count"] == 0
+    assert body["directory"] == os.path.realpath(str(target))
+    assert record["bookkeeping_index_flag"] is False
+
+
+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
+
+    state, enter, leave = _serialization_probe()
+
+    def _slow_index(self, directory, owner=None):
+        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()
+
+    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, (
+        f"{state['max_active']} index jobs ran in parallel — concurrent "
+        "add_directory requests must serialize"
+    )
+
+
+def test_failed_indexing_still_returns_500(tmp_path, monkeypatch):
+    record = {}
+    app = _build_app(tmp_path, monkeypatch, record)
+    target = tmp_path / "docs"
+    target.mkdir()
+
+    def _fail(directory, owner=None):
+        return {"success": False, "message": "boom"}
+
+    monkeypatch.setattr(_FakeRag, "index_personal_documents", staticmethod(_fail))
+
+    client = TestClient(app)
+    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"
+    )

tests/test_issue_description_check.py

@@ -1,86 +0,0 @@
-"""Regression coverage for issue-description label lifecycle events."""
-
-import json
-import shutil
-import subprocess
-from pathlib import Path
-
-import pytest
-
-
-_REPO = Path(__file__).resolve().parent.parent
-_CHECKER = _REPO / ".github" / "scripts" / "check-issue-description.js"
-_WORKFLOW = _REPO / ".github" / "workflows" / "issue-description-check.yml"
-pytestmark = pytest.mark.skipif(not shutil.which("node"), reason="node not on PATH")
-
-
-def _run_closed_issue(action):
-    harness = r"""
-const checkIssueDescription = require(process.argv[1]);
-const action = process.argv[2];
-const calls = [];
-const unexpected = (name) => async () => {
-  throw new Error(`${name} should not be called for a closed issue`);
-};
-
-const github = {
-  rest: {
-    issues: {
-      removeLabel: async (params) => calls.push({ method: 'removeLabel', params }),
-      getLabel: unexpected('getLabel'),
-      addLabels: unexpected('addLabels'),
-      listComments: unexpected('listComments'),
-      createComment: unexpected('createComment'),
-      updateComment: unexpected('updateComment'),
-      deleteComment: unexpected('deleteComment'),
-    },
-  },
-};
-const context = {
-  payload: {
-    action,
-    issue: { number: 42, state: 'closed', body: '', labels: [] },
-  },
-  repo: { owner: 'odysseus-dev', repo: 'odysseus' },
-};
-const core = {
-  warning: unexpected('core.warning'),
-  setFailed: unexpected('core.setFailed'),
-};
-
-checkIssueDescription({ github, context, core })
-  .then(() => process.stdout.write(JSON.stringify(calls)))
-  .catch((error) => {
-    console.error(error);
-    process.exitCode = 1;
-  });
-"""
-    proc = subprocess.run(
-        ["node", "-e", harness, str(_CHECKER), action],
-        capture_output=True,
-        text=True,
-        cwd=str(_REPO),
-        timeout=30,
-    )
-    assert proc.returncode == 0, proc.stderr
-    return json.loads(proc.stdout)
-
-
-def test_workflow_handles_issue_closures():
-    workflow = _WORKFLOW.read_text()
-    assert "types: [opened, edited, reopened, closed]" in workflow
-
-
-@pytest.mark.parametrize("action", ["closed", "edited"])
-def test_closed_issue_only_drops_ready_for_review(action):
-    assert _run_closed_issue(action) == [
-        {
-            "method": "removeLabel",
-            "params": {
-                "owner": "odysseus-dev",
-                "repo": "odysseus",
-                "issue_number": 42,
-                "name": "ready for review",
-            },
-        }
-    ]

tests/test_mcp_dependency_compatibility.py

@@ -1,15 +0,0 @@
-"""Regression coverage for the built-in MCP servers' SDK compatibility line."""
-
-from pathlib import Path
-
-
-REQUIREMENTS = Path(__file__).resolve().parents[1] / "requirements.txt"
-
-
-def test_mcp_requirement_excludes_breaking_v2_sdk():
-    requirements = [
-        line.split("#", 1)[0].strip().replace(" ", "")
-        for line in REQUIREMENTS.read_text(encoding="utf-8").splitlines()
-    ]
-
-    assert "mcp<2" in requirements

tests/test_search_routes_shim.py

@@ -1,11 +0,0 @@
-"""Regression test for the search route shim (slice 2j, #4082/#4071)."""
-
-import importlib
-
-import routes.search_routes as _shim_search  # noqa: F401
-
-
-def test_legacy_and_canonical_search_module_are_same_object():
-    legacy = importlib.import_module("routes.search_routes")
-    canonical = importlib.import_module("routes.search.search_routes")
-    assert legacy is canonical

tests/test_tts_service_enforce_cache_limit.py

@@ -1,97 +0,0 @@
-import os
-import time
-from pathlib import Path
-import pytest
-
-# Adjust the import path if your file is directly in ./services instead of ./services/tts
-from services.tts.tts_service import TTSService
-
-def test_cache_under_limit(tmp_path, monkeypatch):
-    """Test that writing a file under the size limit does not trigger eviction."""
-    # Set a tiny limit: 100 bytes
-    monkeypatch.setenv("ODYSSEUS_TTS_CACHE_MAX_BYTES", "100")
-    
-    # Initialize service with pytest's temporary directory
-    service = TTSService(cache_dir=str(tmp_path))
-    
-    # Write a 40-byte file (under the 100-byte limit)
-    service._put_cache("test_key", b"x" * 40)
-    
-    # Verify the file was written and nothing was deleted
-    files = list(tmp_path.glob("*.*"))
-    assert len(files) == 1
-    assert sum(f.stat().st_size for f in files) == 40
-
-def test_cache_exceeds_limit_triggers_eviction(tmp_path, monkeypatch):
-    """Test that exceeding the limit evicts the oldest files down to 80% capacity."""
-    # Set limit to 100 bytes. 80% target capacity will be 80 bytes.
-    monkeypatch.setenv("ODYSSEUS_TTS_CACHE_MAX_BYTES", "100")
-    service = TTSService(cache_dir=str(tmp_path))
-    
-    # 1. Setup: Manually create two older files (40 bytes each)
-    file1 = tmp_path / "oldest.wav"
-    file2 = tmp_path / "middle.wav"
-    
-    file1.write_bytes(b"a" * 40)
-    file2.write_bytes(b"b" * 40)
-    
-    # Spoof timestamps so file1 is explicitly older than file2
-    now = time.time()
-    os.utime(file1, (now - 100, now - 100)) # 100 seconds ago
-    os.utime(file2, (now - 50, now - 50))   # 50 seconds ago
-    
-    # 2. Action: Write a 3rd file using the service method (40 bytes)
-    # Total cache is now 120 bytes, which exceeds 100.
-    # It should delete oldest (file1) to drop to 80 bytes (which matches the 80% target).
-    service._put_cache("newest", b"c" * 40)
-    
-    # 3. Assertions
-    # The newest file should exist (saved as .wav because it lacks MP3 magic bytes)
-    newest_file = tmp_path / "newest.wav"
-    
-    assert not file1.exists(), "The oldest file should have been evicted."
-    assert file2.exists(), "The middle file should still exist."
-    assert newest_file.exists(), "The newest file should have been saved."
-    
-    # Verify the final directory size is <= 80 bytes
-    total_size = sum(f.stat().st_size for f in tmp_path.glob("*.*"))
-    assert total_size <= 80
-
-def test_cache_limit_disabled(tmp_path, monkeypatch):
-    """Test that setting max bytes to 0 disables eviction."""
-    monkeypatch.setenv("ODYSSEUS_TTS_CACHE_MAX_BYTES", "0")
-    service = TTSService(cache_dir=str(tmp_path))
-    
-    # Write 3 large files that would normally trigger eviction
-    service._put_cache("file1", b"x" * 1000)
-    service._put_cache("file2", b"x" * 1000)
-    service._put_cache("file3", b"x" * 1000)
-    
-    # Ensure nothing was deleted
-    files = list(tmp_path.glob("*.*"))
-    assert len(files) == 3
-    assert sum(f.stat().st_size for f in files) == 3000
-
-def test_cache_eviction_handles_unlink_error_gracefully(tmp_path, monkeypatch):
-    """Test that if unlinking a file fails, _put_cache still succeeds without raising."""
-    service = TTSService(cache_dir=str(tmp_path))
-    service.max_cache_bytes = 50
-
-    # Create a file to evict
-    old_file = tmp_path / "old.wav"
-    old_file.write_bytes(b"x" * 40)
-
-    # Monkeypatch unlink on Path objects to simulate a PermissionError / file-lock failure
-    def mock_unlink(self_path):
-        raise OSError("Permission denied / file locked")
-
-    monkeypatch.setattr(Path, "unlink", mock_unlink)
-
-    # Writing a new file triggers eviction which encounters the mocked unlink error
-    try:
-        service._put_cache("new_key", b"y" * 40)
-    except Exception as e:
-        pytest.fail(f"_put_cache raised an exception during failed eviction: {e}")
-
-    # The new file should still be written successfully
-    assert (tmp_path / "new_key.wav").exists()