fix(personal): bound multi-file upload memory

/api/personal/upload and DELETE /api/personal/file mutated the same
vector and tracking state add/remove/reload serialize on, outside
_index_job_lock and inline on the event loop.

Both now stage async work on the loop, then run the complete transition
(vector writes, disk change, personal_docs_manager update) in one
offloaded critical section under the shared lock, acquired before the
offload so queued requests park on the loop rather than pinning a
threadpool worker.

Adds add-vs-upload and add-vs-file ordering regressions.

routes/chat_routes.py

@@ -21,6 +21,7 @@ from src import agent_runs
 from src.model_context import estimate_tokens
 from src.chat_helpers import coerce_message_and_session
 from src.endpoint_resolver import normalize_base as _normalize_base, build_chat_url
+from src.foreground_model_routing import build_foreground_model_candidates
 from src.session_search import search_session_messages
 from src.prompt_security import untrusted_context_message
 from core.exceptions import SessionNotFoundError
@@ -1399,14 +1400,14 @@ def setup_chat_routes(
             thinking_response = ""
             last_metrics = None
 
-            # Configured fallback chain for the default chat model. Tried in
-            # order if the session's primary model fails before producing
-            # output. Resolved once per request.
-            try:
-                from src.endpoint_resolver import resolve_chat_fallback_candidates
-                _fallback_candidates = resolve_chat_fallback_candidates(owner=_user)
-            except Exception:
-                _fallback_candidates = []
+            # Foreground Chat and Agent requests use one owner-aware policy
+            # boundary. Legacy `default_model_fallbacks` data is not eligible.
+            _foreground_candidates = build_foreground_model_candidates(
+                sess.endpoint_url,
+                sess.model,
+                sess.headers,
+                owner=_user,
+            )
 
             # Send model name early so the frontend can show it during streaming
             _model_suffix = "Research" if effective_do_research else None
@@ -1522,9 +1523,8 @@ def setup_chat_routes(
                 _actual_model = None
                 # ── Chat mode: call stream_llm directly, NO tools, NO document access ──
                 try:
-                    _chat_candidates = [(sess.endpoint_url, sess.model, sess.headers)] + _fallback_candidates
                     async for chunk in stream_llm_with_fallback(
-                        _chat_candidates,
+                        _foreground_candidates,
                         messages,
                         temperature=ctx.preset.temperature,
                         # Respect the preset; 0/unset = let the server decide (no
@@ -1710,7 +1710,7 @@ def setup_chat_routes(
                         disabled_tools=disabled_tools if disabled_tools else None,
                         tool_policy=tool_policy,
                         owner=_user,
-                        fallbacks=_fallback_candidates,
+                        fallbacks=_foreground_candidates[1:],
                         plan_mode=plan_mode,
                         approved_plan=approved_plan or None,
                         workspace=workspace or None,

routes/email_routes.py

@@ -5327,9 +5327,9 @@ def setup_email_routes():
             # Build a candidate chain so a stale session-stored API key
             # (the most common cause of "authentication failed" here)
             # doesn't kill AI Reply outright — fall through to the
-            # user's Utility / Default endpoints AND their configured
-            # fallback chains. Dedupe by url+model so we don't retry
-            # the same broken endpoint.
+            # user's Utility / Default endpoints and the active Utility
+            # fallback chain. The retired default-fallback hook stays empty.
+            # Dedupe by url+model so we don't retry the same broken endpoint.
             from src.llm_core import llm_call_async_with_fallback
             from src.endpoint_resolver import (
                 resolve_utility_fallback_candidates,
@@ -5358,7 +5358,7 @@ def setup_email_routes():
                 _add(_d_url, _d_model, _d_headers)
             except Exception:
                 pass
-            # Configured fallback chains last.
+            # Active Utility fallbacks, then the retired default hook.
             for cand in resolve_utility_fallback_candidates(owner=owner) or []:
                 _add(*cand)
             for cand in resolve_chat_fallback_candidates(owner=owner) or []:

routes/model_routes.py

@@ -46,10 +46,11 @@ _ENDPOINT_SETTING_FIELDS = {
 }
 
 _ENDPOINT_FALLBACK_FIELDS = {
-    "default_model_fallbacks": "Default Model Fallbacks",
     "utility_model_fallbacks": "Utility Model Fallbacks",
     "vision_model_fallbacks":  "Vision Model Fallbacks",
 }
+# `default_model_fallbacks` is intentionally absent. The legacy data remains
+# stored as-is even when an endpoint is removed, but no longer affects routing.
 
 
 def _speech_settings_using_endpoint(settings: dict, ep_id: str) -> list:
@@ -2437,7 +2438,6 @@ def setup_model_routes(model_discovery):
             _user_prefs = _load_for_user(_user) or {}
             ep_id = (_user_prefs.get("default_endpoint_id") or "").strip()
             model = (_user_prefs.get("default_model") or "").strip()
-            _fallbacks = _user_prefs.get("default_model_fallbacks") or []
             # If user has no personal default, fall back to global default
             # But only based on the "share_defaults_with_users" flag
             # (only if share_defaults_with_users is enabled)
@@ -2446,12 +2446,9 @@ def setup_model_routes(model_discovery):
                     ep_id = settings.get("default_endpoint_id", "")
                 if not model:
                     model = settings.get("default_model", "")
-                if not _fallbacks:
-                    _fallbacks = settings.get("default_model_fallbacks") or []
         else:
             ep_id = settings.get("default_endpoint_id", "")
             model = settings.get("default_model", "")
-            _fallbacks = settings.get("default_model_fallbacks") or []
         db = SessionLocal()
         try:
             ep = None
@@ -2466,33 +2463,6 @@ def setup_model_routes(model_discovery):
                 if _user and not _is_admin:
                     ep_q = owner_filter(ep_q, ModelEndpoint, _user)
                 ep = ep_q.first()
-            # Configured fallback chain — when the chosen default endpoint is
-            # gone/disabled, honor the user's configured `default_model_fallbacks`
-            # in order BEFORE arbitrarily grabbing the first enabled endpoint.
-            # (Previously this jumped straight to "first enabled", which is why
-            # deleting/changing the main endpoint silently reassigned the default
-            # chat to some unrelated endpoint instead of the fallback.)
-            if not ep:
-                for entry in _fallbacks:
-                    if not isinstance(entry, dict):
-                        continue
-                    fid = (entry.get("endpoint_id") or "").strip()
-                    if not fid:
-                        continue
-                    cand_q = db.query(ModelEndpoint).filter(
-                        ModelEndpoint.id == fid, ModelEndpoint.is_enabled == True
-                    )
-                    if _user and not _is_admin:
-                        cand_q = owner_filter(cand_q, ModelEndpoint, _user)
-                    cand = cand_q.first()
-                    if cand:
-                        ep = cand
-                        # Use the fallback entry's model. Reset even when empty
-                        # so we don't carry the prior endpoint's stale model onto
-                        # this fallback — the cached-models lookup below then
-                        # fills it from the fallback endpoint.
-                        model = (entry.get("model") or "").strip()
-                        break
             # Last resort: first enabled endpoint owned by THIS user. Do not
             # include null-owner/shared endpoints here: a brand-new user with
             # no explicit default should not auto-open a pending chat using an

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,
@@ -289,54 +332,73 @@ def setup_personal_routes(personal_docs_manager, rag_manager, rag_available):
         total_failed = 0
         uploaded_files = []
 
-        for upload in files:
-            try:
-                file_path, stored_name, safe_name = _unique_personal_upload_path(upload_dir, upload.filename)
-                content_bytes = await upload.read(PERSONAL_UPLOAD_MAX_BYTES + 1)
-                if len(content_bytes) > PERSONAL_UPLOAD_MAX_BYTES:
-                    logger.warning(f"Rejected oversized personal upload: {upload.filename!r}")
-                    total_failed += 1
-                    continue
-                with open(file_path, "wb") as f:
-                    f.write(content_bytes)
-
-                ext = os.path.splitext(safe_name)[1].lower()
-                if ext == ".pdf":
-                    from src.personal_docs import extract_pdf_text
-                    text = extract_pdf_text(file_path)
-                else:
-                    text = content_bytes.decode("utf-8", errors="replace")
-
-                if not text or not text.strip():
-                    total_failed += 1
-                    continue
-
-                # Chunk and index
-                chunks = rag._split_into_chunks(text, chunk_size=500)
-                for i, chunk in enumerate(chunks):
-                    metadata = {
-                        "source": file_path,
-                        "filename": safe_name,
-                        "stored_filename": stored_name,
-                        "directory": upload_dir,
-                        "type": ext,
-                        "chunk_id": i,
-                    }
-                    if user:
-                        metadata["owner"] = user
-                    if rag.add_document(chunk, metadata):
-                        total_indexed += 1
-                    else:
+        # Chunking, embedding and the tracking update are blocking work over the
+        # same vector/tracking state add_directory mutates (#5634). Take the
+        # shared job lock BEFORE offloading so a queued request parks on the loop
+        # instead of pinning a threadpool worker, matching add_directory.
+        # Read and process one capped payload at a time so a multi-file request
+        # cannot retain len(files) * PERSONAL_UPLOAD_MAX_BYTES in memory.
+        async with _index_job_lock:
+            for upload in files:
+                try:
+                    file_path, stored_name, safe_name = _unique_personal_upload_path(
+                        upload_dir, upload.filename
+                    )
+                    content_bytes = await upload.read(PERSONAL_UPLOAD_MAX_BYTES + 1)
+                    if len(content_bytes) > PERSONAL_UPLOAD_MAX_BYTES:
+                        logger.warning(f"Rejected oversized personal upload: {upload.filename!r}")
                         total_failed += 1
+                        continue
 
-                uploaded_files.append(safe_name)
-            except Exception as e:
-                logger.error(f"Failed to upload/index {upload.filename}: {e}")
-                total_failed += 1
+                    def _index_upload():
+                        with open(file_path, "wb") as f:
+                            f.write(content_bytes)
 
-        # Track uploads directory
-        if uploaded_files and hasattr(personal_docs_manager, "add_directory"):
-            personal_docs_manager.add_directory(upload_dir, index=False)
+                        ext = os.path.splitext(safe_name)[1].lower()
+                        if ext == ".pdf":
+                            from src.personal_docs import extract_pdf_text
+                            text = extract_pdf_text(file_path)
+                        else:
+                            text = content_bytes.decode("utf-8", errors="replace")
+
+                        if not text or not text.strip():
+                            return 0, 1, None
+
+                        indexed = 0
+                        failed = 0
+                        chunks = rag._split_into_chunks(text, chunk_size=500)
+                        for i, chunk in enumerate(chunks):
+                            metadata = {
+                                "source": file_path,
+                                "filename": safe_name,
+                                "stored_filename": stored_name,
+                                "directory": upload_dir,
+                                "type": ext,
+                                "chunk_id": i,
+                            }
+                            if user:
+                                metadata["owner"] = user
+                            if rag.add_document(chunk, metadata):
+                                indexed += 1
+                            else:
+                                failed += 1
+                        return indexed, failed, safe_name
+
+                    indexed, failed, uploaded_name = await run_in_threadpool(_index_upload)
+                    total_indexed += indexed
+                    total_failed += failed
+                    if uploaded_name:
+                        uploaded_files.append(uploaded_name)
+                except Exception as e:
+                    logger.error(f"Failed to upload/index {upload.filename}: {e}")
+                    total_failed += 1
+
+            # Same transition, same lock: the tracking update must not land
+            # while another job is mid-write over the same state.
+            if uploaded_files and hasattr(personal_docs_manager, "add_directory"):
+                await run_in_threadpool(
+                    personal_docs_manager.add_directory, upload_dir, index=False
+                )
 
         return {
             "success": True,
@@ -349,38 +411,47 @@ def setup_personal_routes(personal_docs_manager, rag_manager, rag_available):
     async def delete_file_from_rag(filepath: str = Query(...), owner: str = Depends(require_user), _admin: None = Depends(require_admin)):
         """Delete a specific file from RAG index and optionally from disk."""
         try:
-            # Remove chunks from RAG vector store (best-effort)
-            removed = 0
-            rag = _rag()
-            if rag:
-                try:
-                    removed = rag.delete_by_source(filepath)
-                except Exception as e:
-                    logger.warning(f"RAG removal failed for {filepath}: {e}")
+            def _delete_file():
+                # Remove chunks from RAG vector store (best-effort)
+                removed = 0
+                rag = _rag()
+                if rag:
+                    try:
+                        removed = rag.delete_by_source(filepath)
+                    except Exception as e:
+                        logger.warning(f"RAG removal failed for {filepath}: {e}")
 
-            # Delete file from disk if it's in the caller's own uploads dir.
-            # Scope to the per-owner subdir, not the shared uploads root, so one
-            # admin can't delete another user's personal files by path.
-            deleted_from_disk = False
-            try:
-                abs_target = os.path.realpath(filepath)
-                base_abs = os.path.realpath(_personal_upload_dir_for_owner(owner, create=False))
-                in_uploads = (
-                    abs_target == base_abs
-                    or os.path.commonpath([abs_target, base_abs]) == base_abs
-                )
-            except ValueError:
-                # commonpath raises on mixed drives / non-comparable paths
-                in_uploads = False
-            if in_uploads and abs_target != base_abs:
+                # Delete file from disk if it's in the caller's own uploads dir.
+                # Scope to the per-owner subdir, not the shared uploads root, so one
+                # admin can't delete another user's personal files by path.
+                deleted_from_disk = False
                 try:
-                    os.remove(abs_target)
-                    deleted_from_disk = True
-                except FileNotFoundError:
-                    pass  # already gone — race with another request or cleanup
+                    abs_target = os.path.realpath(filepath)
+                    base_abs = os.path.realpath(_personal_upload_dir_for_owner(owner, create=False))
+                    in_uploads = (
+                        abs_target == base_abs
+                        or os.path.commonpath([abs_target, base_abs]) == base_abs
+                    )
+                except ValueError:
+                    # commonpath raises on mixed drives / non-comparable paths
+                    in_uploads = False
+                if in_uploads and abs_target != base_abs:
+                    try:
+                        os.remove(abs_target)
+                        deleted_from_disk = True
+                    except FileNotFoundError:
+                        pass  # already gone — race with another request or cleanup
 
-            # Exclude the file from the listing (persists across restarts)
-            personal_docs_manager.exclude_file(filepath)
+                # Exclude the file from the listing (persists across restarts)
+                personal_docs_manager.exclude_file(filepath)
+                return removed, deleted_from_disk
+
+            # Vector removal, the disk unlink and the exclusion write are one
+            # transition over the same state add_directory mutates (#5634), and
+            # all three block. Take the shared job lock BEFORE offloading, as
+            # add_directory does.
+            async with _index_job_lock:
+                removed, deleted_from_disk = await run_in_threadpool(_delete_file)
 
             return {
                 "success": True,

src/endpoint_resolver.py

@@ -443,28 +443,14 @@ def resolve_endpoint_by_id(
 
 
 def resolve_chat_fallback_candidates(owner: Optional[str] = None) -> list:
-    """Build the configured default-chat fallback chain as a list of
-    (chat_url, model, headers) tuples, skipping any that can't resolve.
+    """Compatibility shim for the retired default-chat fallback chain."""
 
-    The primary model is NOT included — callers prepend their session's
-    current (url, model, headers) so per-session model overrides are honored.
-    """
-    return _resolve_fallback_candidates("default_model_fallbacks", owner=owner)
+    del owner
+    return []
 
 
 def resolve_utility_fallback_candidates(owner: Optional[str] = None) -> list:
     """Configured fallback chain for the Utility model (`utility_model_fallbacks`)."""
-    try:
-        from src.settings import get_user_setting, load_settings
-        settings = load_settings()
-        utility_ep = (get_user_setting("utility_endpoint_id", owner or "", settings.get("utility_endpoint_id", "")) or "").strip()
-        if not utility_ep:
-            utility_chain = get_user_setting("utility_model_fallbacks", owner or "", settings.get("utility_model_fallbacks") or []) or []
-            if utility_chain:
-                return _resolve_fallback_candidates("utility_model_fallbacks", owner=owner)
-            return _resolve_fallback_candidates("default_model_fallbacks", owner=owner)
-    except Exception:
-        pass
     return _resolve_fallback_candidates("utility_model_fallbacks", owner=owner)
 
 

src/foreground_model_routing.py

@@ -0,0 +1,31 @@
+"""Foreground Chat and Agent model-routing policy.
+
+The selected session model is strict by default. Historical
+``default_model_fallbacks`` values remain stored for compatibility, but this
+policy intentionally does not read or migrate them.
+"""
+
+from typing import Any, Dict, Optional
+
+
+def resolve_foreground_fallback_candidates(owner: Optional[str] = None) -> list:
+    """Return fallback candidates for a foreground Chat or Agent request.
+
+    Foreground routing is strict, so no alternate endpoint/model is eligible.
+    ``owner`` is accepted to keep this policy boundary owner-aware.
+    """
+
+    del owner
+    return []
+
+
+def build_foreground_model_candidates(
+    endpoint_url: str,
+    model: str,
+    headers: Optional[Dict[str, Any]] = None,
+    owner: Optional[str] = None,
+) -> list:
+    """Build the ordered candidate list for a foreground request."""
+
+    primary = (endpoint_url, model, headers or {})
+    return [primary] + resolve_foreground_fallback_candidates(owner=owner)

src/llm_core.py

@@ -1949,11 +1949,10 @@ def _dedupe_candidates(candidates):
     """Filter malformed entries and drop a later repeat of an already-seen
     ``(url, model)`` route, preserving order (first occurrence wins).
 
-    The chain is the primary target followed by the configured fallbacks, so a
-    fallback that repeats the session's current model — a common misconfiguration,
-    since callers prepend the live ``(url, model)`` to ``default_model_fallbacks``
-    — would otherwise make the chain re-attempt the very route that just failed:
-    a wasted round-trip plus a spurious ``fallback`` notice for a switch that did
+    The chain is the primary target followed by any caller-authorized
+    fallbacks.  A fallback that repeats the session's current model would
+    otherwise make the chain re-attempt the very route that just failed: a
+    wasted round-trip plus a spurious ``fallback`` notice for a switch that did
     not happen. Headers are not part of the key; the first tuple (with its
     headers) is the one kept.
     """

src/settings.py

@@ -138,14 +138,13 @@ DEFAULT_SETTINGS = {
     # Email replies use email_writing_style instead because greetings,
     # signatures, and mailbox identity rules are medium-specific.
     "document_writing_style": "",
-    # Ordered fallback chain for the default chat model. Each entry is
-    # {"endpoint_id": "...", "model": "..."}. If the primary model fails
-    # before producing output (endpoint offline / errors), the chat
-    # dispatch retries the next entry in order.
+    # Legacy ordered fallback chain for the default chat model. Values remain
+    # stored for compatibility and rollback reference, but model routing no
+    # longer reads this key.
     "default_model_fallbacks": [],
-    # When True, non-admin users inherit global default model/endpoint/fallbacks
-    # when they have no personal defaults. When False, users only use their
-    # personal defaults (no global fallback). Default is False.
+    # When True, non-admin users inherit the global default model/endpoint when
+    # they have no personal defaults. When False, users only use their personal
+    # defaults. Default is False.
     "share_defaults_with_users": False,
     "utility_endpoint_id": "",
     "utility_model": "",

src/task_endpoint.py

@@ -32,7 +32,7 @@ def resolve_task_candidates(
     2. Utility endpoint/model
     3. Default endpoint/model
     4. Utility fallback chain
-    5. Default fallback chain
+    5. Retired default-fallback compatibility hook (currently empty)
     """
     candidates = []
 

static/index.html

@@ -1504,7 +1504,7 @@
                 <span class="adm-model-logo" id="set-defaultModelSelect-logo" style="display:inline-flex;align-items:center;justify-content:center;width:18px;height:18px;flex-shrink:0;opacity:0.9;color:var(--fg);"></span>
                 <select id="set-defaultModelSelect" class="settings-select"></select>
               </div>
-              <div class="settings-row" style="align-items:flex-start;">
+              <div class="settings-row" style="align-items:flex-start;" hidden>
                 <label class="settings-label" style="margin-top:6px;">Fallbacks</label>
                 <div style="flex:1;display:flex;flex-direction:column;gap:6px;">
                   <div id="set-defaultFallbacks" class="settings-fallbacks"></div>

static/js/settings.js

@@ -448,7 +448,7 @@ async function initDefaultChat() {
   var fbContainer = el('set-defaultFallbacks');
   var addFbBtn = el('set-defaultAddFallback');
   var _endpoints = [];
-  var _fallbacks = []; // [{endpoint_id, model}] — tried in order if primary fails
+  var _fallbacks = []; // Hidden legacy DOM hook; stored values are not loaded or saved.
 
   function enabledEndpoints() {
     return _endpoints.filter(function(e) { return e.is_enabled; });
@@ -534,11 +534,6 @@ async function initDefaultChat() {
     var settings = await res.json();
     if (settings.default_endpoint_id) epSel.value = settings.default_endpoint_id;
     refreshModels(settings.default_model || '');
-    _fallbacks = Array.isArray(settings.default_model_fallbacks)
-      ? settings.default_model_fallbacks.map(function(f) {
-          return { endpoint_id: (f && f.endpoint_id) || '', model: (f && f.model) || '' };
-        })
-      : [];
     renderFallbacks();
   } catch (e) { console.warn('Failed to load default chat settings', e); }
 
@@ -547,13 +542,11 @@ async function initDefaultChat() {
 
   async function saveDefault() {
     try {
-      var clean = _fallbacks.filter(function(f) { return f.endpoint_id && f.model; });
       await fetch('/api/auth/settings', { method: 'POST', credentials: 'same-origin',
         headers: { 'Content-Type': 'application/json' },
         body: JSON.stringify({
           default_endpoint_id: epSel.value,
-          default_model: modelSel.value,
-          default_model_fallbacks: clean
+          default_model: modelSel.value
         })
       });
       msg.textContent = 'Saved'; msg.style.color = 'var(--fg)';

tests/test_add_directory_event_loop.py

@@ -0,0 +1,394 @@
+"""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}
+
+    def _split_into_chunks(self, text, chunk_size=500):
+        return [text]
+
+    def add_document(self, chunk, metadata):
+        self._record["add_document_thread"] = threading.get_ident()
+        return True
+
+    def delete_by_source(self, filepath):
+        self._record["delete_thread"] = threading.get_ident()
+        return 1
+
+
+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 exclude_file(self, filepath):
+        self._record["exclude_thread"] = threading.get_ident()
+
+
+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_add_and_upload_serialize(tmp_path, monkeypatch):
+    """#5634 follow-up: POST /upload writes chunks into the vector store and then
+    calls personal_docs_manager.add_directory — the same vector/tracking state
+    add_directory mutates. It must hold the SAME job lock, or an upload landing
+    mid-add interleaves two writers over unsynchronized state."""
+    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_add_document(self, chunk, metadata):
+        self._record["add_document_thread"] = threading.get_ident()
+        enter(); time.sleep(0.25); leave()
+        return True
+
+    monkeypatch.setattr(_FakeRag, "index_personal_documents", _slow_index)
+    monkeypatch.setattr(_FakeRag, "add_document", _slow_add_document)
+
+    record = {}
+    app = _build_app(tmp_path, monkeypatch, record)
+    monkeypatch.setattr(personal_routes, "UPLOADS_DIR", str(tmp_path / "uploads"))
+    monkeypatch.setattr(personal_routes, "require_privilege", lambda request, key: "tester")
+    (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/upload", files={"files": ("a.txt", b"hello world", "text/plain")}),
+        )
+
+    assert all(r.status_code == 200 for r in results)
+    # The test coroutine runs on the event loop, so this IS the loop thread.
+    assert record["add_document_thread"] != threading.get_ident(), (
+        "rag.add_document ran on the event loop thread — chunk writes block "
+        "every other request for the duration of the upload"
+    )
+    assert state["max_active"] == 1, (
+        f"{state['max_active']} add/upload critical sections overlapped — "
+        "upload must hold the same index job lock as add"
+    )
+
+
+async def test_upload_processes_each_payload_before_reading_the_next(tmp_path, monkeypatch):
+    """A multi-file upload must retain at most one capped payload at a time."""
+    from starlette.datastructures import UploadFile as StarletteUploadFile
+
+    reads = []
+    original_read = StarletteUploadFile.read
+
+    async def _recording_read(upload, size=-1):
+        reads.append(upload.filename)
+        return await original_read(upload, size)
+
+    def _record_first_index(self, chunk, metadata):
+        self._record.setdefault("reads_at_first_index", len(reads))
+        return True
+
+    monkeypatch.setattr(StarletteUploadFile, "read", _recording_read)
+    monkeypatch.setattr(_FakeRag, "add_document", _record_first_index)
+
+    record = {}
+    app = _build_app(tmp_path, monkeypatch, record)
+    monkeypatch.setattr(personal_routes, "UPLOADS_DIR", str(tmp_path / "uploads"))
+    monkeypatch.setattr(personal_routes, "require_privilege", lambda request, key: "tester")
+
+    files = [
+        ("files", ("a.txt", b"alpha", "text/plain")),
+        ("files", ("b.txt", b"bravo", "text/plain")),
+        ("files", ("c.txt", b"charlie", "text/plain")),
+    ]
+    async with _async_client(app) as ac:
+        response = await ac.post("/api/personal/upload", files=files)
+
+    assert response.status_code == 200
+    assert response.json()["uploaded"] == ["a.txt", "b.txt", "c.txt"]
+    assert reads == ["a.txt", "b.txt", "c.txt"]
+    assert record["reads_at_first_index"] == 1, (
+        "all upload bodies were retained before worker processing began"
+    )
+
+
+async def test_add_and_delete_file_serialize(tmp_path, monkeypatch):
+    """#5634 follow-up: DELETE /file removes chunks from the vector store and
+    calls personal_docs_manager.exclude_file. Both mutate state add_directory
+    also touches, so the delete must hold the SAME job lock as add."""
+    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_delete(self, filepath):
+        self._record["delete_thread"] = threading.get_ident()
+        enter(); time.sleep(0.25); leave()
+        return 1
+
+    monkeypatch.setattr(_FakeRag, "index_personal_documents", _slow_index)
+    monkeypatch.setattr(_FakeRag, "delete_by_source", _slow_delete)
+
+    record = {}
+    app = _build_app(tmp_path, monkeypatch, record)
+    monkeypatch.setattr(personal_routes, "UPLOADS_DIR", str(tmp_path / "uploads"))
+    (tmp_path / "docs_a").mkdir()
+    doomed = tmp_path / "doomed.txt"
+    doomed.write_text("bye")
+
+    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/file", params={"filepath": str(doomed)}),
+        )
+
+    assert all(r.status_code == 200 for r in results)
+    assert record["delete_thread"] != threading.get_ident(), (
+        "rag.delete_by_source ran on the event loop thread"
+    )
+    assert state["max_active"] == 1, (
+        f"{state['max_active']} add/delete critical sections overlapped — "
+        "delete 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_foreground_model_routing.py

@@ -0,0 +1,277 @@
+"""Regression coverage for strict foreground model selection."""
+
+import asyncio
+import json
+from types import SimpleNamespace
+
+import pytest
+import src.agent_loop as agent_loop
+import src.endpoint_resolver as endpoint_resolver
+import src.foreground_model_routing as foreground_model_routing
+import routes.chat_routes as chat_routes
+from src.foreground_model_routing import (
+    build_foreground_model_candidates,
+    resolve_foreground_fallback_candidates,
+)
+
+
+def _collect(gen):
+    async def _run():
+        return [chunk async for chunk in gen]
+
+    return asyncio.run(_run())
+
+
+class _EmptyQuery:
+    def filter(self, *args, **kwargs):
+        return self
+
+    def order_by(self, *args, **kwargs):
+        return self
+
+    def first(self):
+        return None
+
+
+class _EmptyDb:
+    def query(self, *args, **kwargs):
+        return _EmptyQuery()
+
+    def close(self):
+        return None
+
+
+class _RouteRequest:
+    def __init__(self, mode):
+        self.headers = {}
+        self.app = SimpleNamespace(state=SimpleNamespace(auth_manager=None))
+        self._form = {
+            "message": "hello",
+            "session": "session-1",
+            "mode": mode,
+            "compare_mode": "true",
+        }
+
+    async def form(self):
+        return self._form
+
+
+def _chat_stream_endpoint(monkeypatch, mode, captured):
+    session = SimpleNamespace(
+        endpoint_url="https://selected.example/v1",
+        model="selected-model",
+        headers={"Authorization": "Bearer selected"},
+        name="test",
+        history=[],
+        add_message=lambda message: None,
+    )
+    session_manager = SimpleNamespace(
+        get_session=lambda session_id: session,
+        save_sessions=lambda: None,
+    )
+    context = SimpleNamespace(
+        user="alice",
+        messages=[{"role": "user", "content": "hello"}],
+        preprocessed=SimpleNamespace(attachment_meta=[]),
+        auto_opened_docs=[],
+        rag_sources=[],
+        web_sources=[],
+        used_memories=[],
+        uploaded_files=[],
+        uprefs={},
+        was_compacted=False,
+        context_trimmed=False,
+        context_length=4096,
+        context_messages_before_trim=1,
+        context_messages_after_trim=1,
+        context_tokens_before_trim=10,
+        context_tokens_after_trim=10,
+        preset=SimpleNamespace(temperature=0.2, max_tokens=128, character_name=None),
+    )
+
+    async def fake_build_context(*args, **kwargs):
+        return context
+
+    async def fake_chat_stream(candidates, messages, **kwargs):
+        captured["chat"] = candidates
+        yield f'data: {json.dumps({"delta": "done"})}\n\n'
+        yield "data: [DONE]\n\n"
+
+    async def fake_agent_stream(endpoint_url, model, messages, **kwargs):
+        captured["agent"] = {
+            "primary": (endpoint_url, model, kwargs.get("headers")),
+            "fallbacks": kwargs.get("fallbacks"),
+        }
+        yield f'data: {json.dumps({"delta": "done"})}\n\n'
+        yield "data: [DONE]\n\n"
+
+    monkeypatch.setattr(chat_routes, "coerce_message_and_session", lambda *args, **kwargs: ("hello", "session-1"))
+    monkeypatch.setattr(chat_routes, "_verify_session_owner", lambda *args, **kwargs: None)
+    monkeypatch.setattr(chat_routes, "effective_user", lambda request: "alice")
+    monkeypatch.setattr(chat_routes, "_clear_orphaned_session_endpoint", lambda *args, **kwargs: False)
+    monkeypatch.setattr(chat_routes, "_recover_empty_session_model", lambda *args, **kwargs: False)
+    monkeypatch.setattr(chat_routes, "_enforce_chat_privileges", lambda *args, **kwargs: None)
+    monkeypatch.setattr(chat_routes, "resolve_session_auth", lambda *args, **kwargs: None)
+    monkeypatch.setattr(chat_routes, "get_session_mode", lambda session_id: "chat")
+    monkeypatch.setattr(chat_routes, "set_session_mode", lambda *args, **kwargs: None)
+    monkeypatch.setattr(chat_routes, "build_chat_context", fake_build_context)
+    monkeypatch.setattr(chat_routes, "SessionLocal", _EmptyDb)
+    monkeypatch.setattr(chat_routes, "_is_image_generation_session", lambda *args, **kwargs: False)
+    monkeypatch.setattr(chat_routes, "stream_llm_with_fallback", fake_chat_stream)
+    monkeypatch.setattr(chat_routes, "stream_agent_loop", fake_agent_stream)
+    monkeypatch.setattr(chat_routes, "save_assistant_response", lambda *args, **kwargs: None)
+    monkeypatch.setattr(chat_routes, "run_post_response_tasks", lambda *args, **kwargs: None)
+    monkeypatch.setattr(chat_routes, "estimate_tokens", lambda messages: 10)
+    monkeypatch.setattr(
+        endpoint_resolver,
+        "resolve_chat_fallback_candidates",
+        lambda owner=None: [("https://legacy.example/v1", "legacy-model", {})],
+    )
+
+    import src.settings as settings
+
+    monkeypatch.setattr(
+        settings,
+        "get_setting",
+        lambda key, default=None: default,
+    )
+    monkeypatch.setattr(
+        settings,
+        "get_user_setting",
+        lambda key, owner="", default=None: (
+            [{"endpoint_id": "legacy", "model": "legacy-model"}]
+            if key == "default_model_fallbacks"
+            else default
+        ),
+    )
+
+    router = chat_routes.setup_chat_routes(
+        session_manager,
+        SimpleNamespace(),
+        SimpleNamespace(),
+        SimpleNamespace(),
+        SimpleNamespace(),
+        SimpleNamespace(),
+    )
+    return next(route.endpoint for route in router.routes if route.path == "/api/chat_stream")
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("mode", ["chat", "agent"])
+async def test_chat_stream_route_keeps_selected_model_strict_with_legacy_data(monkeypatch, mode):
+    captured = {}
+    endpoint = _chat_stream_endpoint(monkeypatch, mode, captured)
+
+    response = await endpoint(_RouteRequest(mode))
+    async for _ in response.body_iterator:
+        pass
+
+    selected = (
+        "https://selected.example/v1",
+        "selected-model",
+        {"Authorization": "Bearer selected"},
+    )
+    if mode == "chat":
+        assert captured == {"chat": [selected]}
+    else:
+        assert captured == {"agent": {"primary": selected, "fallbacks": []}}
+
+
+def test_candidate_builder_appends_only_policy_authorized_fallbacks(monkeypatch):
+    """Chat and Agent share the same candidate-building policy boundary."""
+
+    authorized = [("https://opt-in.example/v1", "opt-in-model", {})]
+    monkeypatch.setattr(
+        foreground_model_routing,
+        "resolve_foreground_fallback_candidates",
+        lambda owner=None: authorized,
+    )
+
+    assert build_foreground_model_candidates(
+        "https://selected.example/v1",
+        "selected-model",
+        {"Authorization": "Bearer selected"},
+        owner="alice",
+    ) == [
+        ("https://selected.example/v1", "selected-model", {"Authorization": "Bearer selected"}),
+        *authorized,
+    ]
+
+
+def test_strict_policy_builds_only_the_selected_chat_candidate():
+    candidates = build_foreground_model_candidates(
+        "https://selected.example/v1",
+        "selected-model",
+        {"Authorization": "Bearer selected"},
+        owner="alice",
+    )
+
+    assert candidates == [
+        ("https://selected.example/v1", "selected-model", {"Authorization": "Bearer selected"})
+    ]
+
+
+def test_legacy_chat_resolver_is_disconnected():
+    assert endpoint_resolver.resolve_chat_fallback_candidates(owner="alice") == []
+
+
+def test_utility_resolver_does_not_inherit_legacy_chat_fallbacks(monkeypatch):
+    seen_keys = []
+
+    def fake_resolve(setting_key, owner=None):
+        seen_keys.append((setting_key, owner))
+        return [("https://utility.example/v1", "utility-model", {})]
+
+    monkeypatch.setattr(endpoint_resolver, "_resolve_fallback_candidates", fake_resolve)
+
+    assert endpoint_resolver.resolve_utility_fallback_candidates(owner="alice") == [
+        ("https://utility.example/v1", "utility-model", {})
+    ]
+    assert seen_keys == [("utility_model_fallbacks", "alice")]
+
+
+def test_multi_round_agent_uses_only_selected_model(monkeypatch):
+    """Every Agent round receives only the selected foreground candidate."""
+
+    seen_candidates = []
+    round_number = 0
+
+    monkeypatch.setattr(agent_loop, "get_setting", lambda key, default=None: default)
+    monkeypatch.setattr(agent_loop, "get_mcp_manager", lambda: None)
+    monkeypatch.setattr(agent_loop, "estimate_tokens", lambda *args, **kwargs: 10)
+
+    async def fake_stream(candidates, messages, **kwargs):
+        nonlocal round_number
+        round_number += 1
+        seen_candidates.append([(url, model) for url, model, _headers in candidates])
+        if round_number == 1:
+            call = {"name": "bash", "arguments": json.dumps({"command": "printf ok"})}
+            yield f'data: {json.dumps({"type": "tool_calls", "calls": [call]})}\n\n'
+        else:
+            yield f'data: {json.dumps({"delta": "done"})}\n\n'
+        yield "data: [DONE]\n\n"
+
+    async def fake_execute(block, *args, **kwargs):
+        return "bash", {"output": "ok", "exit_code": 0}
+
+    monkeypatch.setattr(agent_loop, "stream_llm_with_fallback", fake_stream)
+    monkeypatch.setattr(agent_loop, "execute_tool_block", fake_execute)
+
+    fallbacks = resolve_foreground_fallback_candidates(owner="alice")
+    chunks = _collect(
+        agent_loop.stream_agent_loop(
+            "https://selected.example/v1",
+            "selected-model",
+            [{"role": "user", "content": "Run one tool and report back."}],
+            max_rounds=3,
+            relevant_tools={"bash"},
+            fallbacks=fallbacks,
+            _is_teacher_run=True,
+        )
+    )
+
+    assert seen_candidates == [
+        [("https://selected.example/v1", "selected-model")],
+        [("https://selected.example/v1", "selected-model")],
+    ]
+    assert any('"delta": "done"' in chunk for chunk in chunks)

tests/test_legacy_default_fallback_ui.py

@@ -0,0 +1,29 @@
+"""The retired default fallback editor must not imply active routing."""
+
+from pathlib import Path
+
+from bs4 import BeautifulSoup
+
+
+_REPO = Path(__file__).resolve().parents[1]
+
+
+def test_legacy_default_fallback_editor_is_hidden():
+    soup = BeautifulSoup(
+        (_REPO / "static" / "index.html").read_text(encoding="utf-8"),
+        "html.parser",
+    )
+    editor = soup.find(id="set-defaultFallbacks")
+
+    assert editor is not None
+    assert editor.find_parent(class_="settings-row").has_attr("hidden")
+
+
+def test_default_model_save_does_not_rewrite_legacy_fallbacks():
+    source = (_REPO / "static" / "js" / "settings.js").read_text(encoding="utf-8")
+    start = source.index("async function initDefaultChat()")
+    end = source.index("/* ── Utility Model ── */", start)
+    default_chat_source = source[start:end]
+
+    assert "settings.default_model_fallbacks" not in default_chat_source
+    assert "default_model_fallbacks:" not in default_chat_source

tests/test_model_defaults.py

@@ -144,10 +144,10 @@ def test_get_default_chat_user_no_prefs_share_disabled_resolves_nothing(monkeypa
     assert test_data["model"] == "", "Should get empty model"
 
 
-def test_get_default_chat_user_no_prefs_share_enabled_resolves_global_defaults_fallbacks(monkeypatch):
+def test_get_default_chat_user_no_prefs_share_enabled_resolves_global_defaults(monkeypatch):
     """
     Non-admin user without personal preferences should resolve to global
-    defaults for ep_id, model, and fallbacks when share_defaults_with_users is enabled.
+    defaults for ep_id and model when share_defaults_with_users is enabled.
     """
 
     test_data = _run_get_default_chat_test(monkeypatch, share_defaults_enabled=True)
@@ -158,16 +158,45 @@ def test_get_default_chat_user_no_prefs_share_enabled_resolves_global_defaults_f
     assert test_data["endpoint_id"] == "global-ep-123", \
         "Should get global endpoint_id"
 
-def test_get_default_chat_user_no_prefs_share_enabled_resolves_global_defaults(monkeypatch):
+def test_get_default_chat_does_not_read_legacy_fallbacks(monkeypatch):
     """
-    Non-admin user without personal preferences should resolve to global
-    defaults for ep_id, model, and fallbacks when share_defaults_with_users is enabled.
+    The preserved legacy list must not influence default model resolution.
     """
 
-    test_data = _run_get_default_chat_test(monkeypatch, share_defaults_enabled=True, second_endpoint_only=True)
+    class LegacyReadGuard(dict):
+        def get(self, key, default=None):
+            if key == "default_model_fallbacks":
+                raise AssertionError("legacy fallback list was read")
+            return super().get(key, default)
 
-    assert test_data["model"] == "qwen-3.6", \
-        "model should be resolved from global default_model"
+    guarded_settings = LegacyReadGuard({
+        "default_endpoint_id": "global-ep-123",
+        "default_model": "qwen-3.6",
+        "default_model_fallbacks": [
+            {"endpoint_id": "fallback-ep", "model": "fallback-model"}
+        ],
+        "share_defaults_with_users": True,
+    })
+    monkeypatch.setattr(model_routes, "_load_settings", lambda: guarded_settings)
+    monkeypatch.setattr(prefs_routes, "_load_for_user", lambda user: LegacyReadGuard({}))
 
-    assert test_data["endpoint_id"] == "fallback-ep", \
-        "Should get global endpoint_id"
+    fake_auth_manager = MagicMock()
+    fake_auth_manager.is_admin = lambda user: False
+    endpoint = _FakeEndpoint(
+        id="global-ep-123",
+        base_url="http://global-endpoint:8000/v1",
+        is_enabled=True,
+    )
+    fake_db = _make_db_session([endpoint], user="regular_user")
+    monkeypatch.setattr(model_routes, "SessionLocal", lambda: fake_db)
+    monkeypatch.setattr(model_routes, "_normalize_base", lambda url: url)
+    monkeypatch.setattr(model_routes, "build_chat_url", lambda base: f"{base}/chat")
+
+    router = model_routes.setup_model_routes(model_discovery=None)
+    get_default_chat = _get_default_chat_route(router)
+    fake_request = _make_request(user="regular_user", auth_manager=fake_auth_manager)
+
+    test_data = get_default_chat(fake_request)
+
+    assert test_data["endpoint_id"] == "global-ep-123"
+    assert test_data["model"] == "qwen-3.6"

tests/test_model_routes.py

@@ -90,7 +90,7 @@ def test_clear_speech_endpoint_settings_resets_tts_and_stt():
     }
 
 
-def test_endpoint_cleanup_removes_primary_and_fallback_references():
+def test_endpoint_cleanup_preserves_legacy_default_fallback_data():
     settings = {
         "default_endpoint_id": "dead",
         "default_model": "primary",
@@ -106,14 +106,12 @@ def test_endpoint_cleanup_removes_primary_and_fallback_references():
 
     assert _endpoint_settings_using_endpoint(settings, "dead", include_speech=True) == [
         "Default Model",
-        "Default Model Fallbacks",
         "Utility Model Fallbacks",
         "Vision Model Fallbacks",
         "Speech to Text",
     ]
     assert _clear_endpoint_settings_for_endpoint(settings, "dead", include_speech=True) == [
         "Default Model",
-        "Default Model Fallbacks",
         "Utility Model Fallbacks",
         "Vision Model Fallbacks",
         "Speech to Text",
@@ -121,6 +119,7 @@ def test_endpoint_cleanup_removes_primary_and_fallback_references():
     assert settings["default_endpoint_id"] == ""
     assert settings["default_model"] == ""
     assert settings["default_model_fallbacks"] == [
+        {"endpoint_id": "dead", "model": "fallback-a"},
         {"endpoint_id": "keep", "model": "fallback-b"},
     ]
     assert settings["utility_model_fallbacks"] == []
@@ -129,7 +128,7 @@ def test_endpoint_cleanup_removes_primary_and_fallback_references():
     assert settings["stt_model"] == "base"
 
 
-def test_endpoint_cleanup_updates_scoped_and_legacy_user_prefs():
+def test_endpoint_cleanup_updates_active_scoped_prefs_but_preserves_legacy_data():
     scoped = {
         "_users": {
             "alice": {
@@ -154,8 +153,10 @@ def test_endpoint_cleanup_updates_scoped_and_legacy_user_prefs():
     legacy = {
         "default_model_fallbacks": [{"endpoint_id": "dead", "model": "chat"}],
     }
-    assert _clear_user_pref_endpoint_refs(legacy, "dead") == 1
-    assert legacy["default_model_fallbacks"] == []
+    assert _clear_user_pref_endpoint_refs(legacy, "dead") == 0
+    assert legacy["default_model_fallbacks"] == [
+        {"endpoint_id": "dead", "model": "chat"}
+    ]
 
 
 # ── _default_endpoint_needs_assignment (add-endpoint auto-default) ──