Merge pull request #476 from christianlouis/copilot/add-multi-user-support

Fix MagicMock leaking into SQLAlchemy INSERT in filename preservation tests

.env.demo

@@ -129,6 +129,18 @@ ADMIN_USERNAME=admin
 ADMIN_PASSWORD=your_secure_password
 ADMIN_GROUP_NAME=admin
 
+# **Multi-User Mode**
+# When enabled, each user has their own document space with isolated uploads,
+# search, and file management. Requires AUTH_ENABLED=true.
+MULTI_USER_ENABLED=false
+# Default upload limit per user per day (0 = unlimited)
+DEFAULT_DAILY_UPLOAD_LIMIT=0
+# Show unowned documents (owner_id=NULL) to all users (true) or only admins (false)
+UNOWNED_DOCS_VISIBLE_TO_ALL=true
+# Auto-assign this owner ID to documents ingested without a session (e.g. IMAP, API)
+# Leave empty/unset to keep them unowned until claimed.
+# DEFAULT_OWNER_ID=
+
 # **OpenID Connect/Authentik Settings**
 AUTHENTIK_CLIENT_ID=<yourAuthentikAppClientID>
 AUTHENTIK_CLIENT_SECRET=<yourAuthentikClientSecret>

app/api/files.py

@@ -28,6 +28,7 @@ from app.utils.file_queries import apply_status_filter
 from app.utils.file_status import get_files_processing_status
 from app.utils.filename_utils import sanitize_filename
 from app.utils.input_validation import validate_search_query, validate_sort_field, validate_sort_order
+from app.utils.user_scope import apply_owner_filter, get_current_owner_id
 
 # Set up logging
 logger = logging.getLogger(__name__)
@@ -99,8 +100,9 @@ def list_files_api(
     validate_sort_order(sort_order)
     search = validate_search_query(search)
 
-    # Start with base query
+    # Start with base query, scoped to the current user in multi-user mode
     query = db.query(FileRecord)
+    query = apply_owner_filter(query, request)
 
     # Apply search filter
     if search:
@@ -241,8 +243,10 @@ def get_file_details(request: Request, file_id: int, db: DbSession):
     """
     Get detailed information about a specific file including processing history.
     """
-    # Find the file record
+    # Find the file record, scoped to the current user in multi-user mode
-    file_record = db.query(FileRecord).filter(FileRecord.id == file_id).first()
+    query = db.query(FileRecord).filter(FileRecord.id == file_id)
+    query = apply_owner_filter(query, request)
+    file_record = query.first()
 
     if not file_record:
         raise HTTPException(status_code=404, detail=f"File record with ID {file_id} not found")
@@ -300,8 +304,10 @@ def delete_file_record(request: Request, file_id: int, db: DbSession):
         raise HTTPException(status_code=403, detail="File deletion is disabled in the configuration")
 
     try:
-        # Find the file record
+        # Find the file record, scoped to the current user in multi-user mode
-        file_record = db.query(FileRecord).filter(FileRecord.id == file_id).first()
+        query = db.query(FileRecord).filter(FileRecord.id == file_id)
+        query = apply_owner_filter(query, request)
+        file_record = query.first()
 
         if not file_record:
             raise HTTPException(status_code=404, detail=f"File record with ID {file_id} not found")
@@ -1286,6 +1292,9 @@ async def ui_upload(request: Request, db: DbSession, file: UploadFile = File(...
     mime_type, _ = mimetypes.guess_type(target_path)
     file_ext = os.path.splitext(target_path)[1].lower()
 
+    # Determine the owner_id for multi-user document isolation
+    upload_owner_id = get_current_owner_id(request) if settings.multi_user_enabled else None
+
     # Check if it's a PDF by extension or MIME type
     is_pdf = file_ext == ".pdf" or mime_type == "application/pdf"
 
@@ -1310,7 +1319,7 @@ async def ui_upload(request: Request, db: DbSession, file: UploadFile = File(...
             task_ids = []
             for split_file in split_files:
                 split_filename = os.path.basename(split_file)
-                task = process_document.delay(split_file, original_filename=split_filename)
+                task = process_document.delay(split_file, original_filename=split_filename, owner_id=upload_owner_id)
                 task_ids.append(task.id)
                 logger.info(f"Enqueued split PDF part for processing: {split_file}")
 
@@ -1333,7 +1342,7 @@ async def ui_upload(request: Request, db: DbSession, file: UploadFile = File(...
 
     if is_pdf and not should_split:
         # If it's a PDF, process directly
-        task = process_document.delay(target_path, original_filename=safe_filename)
+        task = process_document.delay(target_path, original_filename=safe_filename, owner_id=upload_owner_id)
         logger.info(f"Enqueued PDF for processing: {target_path}")
     elif mime_type in IMAGE_MIME_TYPES or file_ext in {
         ".jpg",
@@ -1347,16 +1356,16 @@ async def ui_upload(request: Request, db: DbSession, file: UploadFile = File(...
         ".svg",
     }:
         # If it's an image, convert to PDF first
-        task = convert_to_pdf.delay(target_path, original_filename=safe_filename)
+        task = convert_to_pdf.delay(target_path, original_filename=safe_filename, owner_id=upload_owner_id)
         logger.info(f"Enqueued image for PDF conversion: {target_path}")
     elif mime_type in ALLOWED_MIME_TYPES or file_ext in ALLOWED_EXTENSIONS:
         # Office document, HTML, Markdown, or other Gotenberg-supported format
-        task = convert_to_pdf.delay(target_path, original_filename=safe_filename)
+        task = convert_to_pdf.delay(target_path, original_filename=safe_filename, owner_id=upload_owner_id)
         logger.info(f"Enqueued document for PDF conversion: {target_path}")
     else:
         # For any other file type, attempt conversion but log a warning
         logger.warning(f"Unsupported MIME type {mime_type} for {target_path}, attempting conversion")
-        task = convert_to_pdf.delay(target_path, original_filename=safe_filename)
+        task = convert_to_pdf.delay(target_path, original_filename=safe_filename, owner_id=upload_owner_id)
 
     # Check for exact duplicates (same SHA-256 hash) before returning.
     # This gives the caller an immediate warning without waiting for the pipeline.
@@ -1394,3 +1403,145 @@ async def ui_upload(request: Request, db: DbSession, file: UploadFile = File(...
     if exact_duplicate_warning:
         response["duplicate_warning"] = exact_duplicate_warning
     return response
+
+
+# ---------------------------------------------------------------------------
+# Document ownership / claim endpoints
+# ---------------------------------------------------------------------------
+
+
+@router.post("/files/{file_id}/claim")
+@require_login
+def claim_file(request: Request, file_id: int, db: DbSession):
+    """
+    Claim an unowned document for the current user.
+
+    Only documents with ``owner_id IS NULL`` can be claimed.  The requesting
+    user's identifier is written into ``owner_id``.  In single-user mode
+    the endpoint is a no-op (returns the file unchanged).
+    """
+    if not settings.multi_user_enabled:
+        raise HTTPException(status_code=400, detail="Multi-user mode is not enabled")
+
+    owner_id = get_current_owner_id(request)
+    if owner_id is None:
+        raise HTTPException(status_code=401, detail="Authentication required to claim a document")
+
+    file_record = db.query(FileRecord).filter(FileRecord.id == file_id).first()
+    if not file_record:
+        raise HTTPException(status_code=404, detail=f"File record with ID {file_id} not found")
+
+    if file_record.owner_id is not None:
+        if file_record.owner_id == owner_id:
+            return {"status": "already_owned", "message": "You already own this document", "file_id": file_id}
+        raise HTTPException(status_code=403, detail="This document is already owned by another user")
+
+    file_record.owner_id = owner_id
+    try:
+        db.commit()
+    except Exception as e:
+        db.rollback()
+        logger.exception(f"Error claiming file {file_id}: {e}")
+        raise HTTPException(status_code=500, detail="Failed to claim document")
+
+    logger.info(f"File {file_id} claimed by user '{owner_id}'")
+    return {"status": "success", "message": "Document claimed successfully", "file_id": file_id, "owner_id": owner_id}
+
+
+@router.post("/files/bulk-claim")
+@require_login
+def bulk_claim_files(request: Request, file_ids: list[int], db: DbSession):
+    """
+    Claim multiple unowned documents for the current user.
+
+    Only documents with ``owner_id IS NULL`` will be claimed.  Documents
+    already owned (by anyone) are skipped and reported in ``skipped``.
+    """
+    if not settings.multi_user_enabled:
+        raise HTTPException(status_code=400, detail="Multi-user mode is not enabled")
+
+    owner_id = get_current_owner_id(request)
+    if owner_id is None:
+        raise HTTPException(status_code=401, detail="Authentication required to claim documents")
+
+    file_records = db.query(FileRecord).filter(FileRecord.id.in_(file_ids)).all()
+    if not file_records:
+        raise HTTPException(status_code=404, detail="No files found with the provided IDs")
+
+    claimed = []
+    skipped = []
+    for rec in file_records:
+        if rec.owner_id is None:
+            rec.owner_id = owner_id
+            claimed.append(rec.id)
+        else:
+            skipped.append({"file_id": rec.id, "reason": "already owned"})
+
+    try:
+        db.commit()
+    except Exception as e:
+        db.rollback()
+        logger.exception(f"Error during bulk claim: {e}")
+        raise HTTPException(status_code=500, detail="Failed to claim documents")
+
+    logger.info(f"Bulk claim by '{owner_id}': claimed={claimed}, skipped={[s['file_id'] for s in skipped]}")
+    return {
+        "status": "success",
+        "claimed_count": len(claimed),
+        "claimed_ids": claimed,
+        "skipped": skipped,
+        "owner_id": owner_id,
+    }
+
+
+@router.post("/files/assign-owner")
+@require_login
+def assign_owner(request: Request, db: DbSession, owner_id: str = Query(...), file_ids: list[int] | None = None):
+    """
+    Admin-only: assign an owner to documents.
+
+    If ``file_ids`` is provided, only those files are updated.  If omitted,
+    **all** currently unowned documents (``owner_id IS NULL``) are assigned
+    to the given ``owner_id``.
+    """
+    if not settings.multi_user_enabled:
+        raise HTTPException(status_code=400, detail="Multi-user mode is not enabled")
+
+    user = request.session.get("user")
+    if not isinstance(user, dict) or not user.get("is_admin"):
+        raise HTTPException(status_code=403, detail="Only admins can assign document owners")
+
+    if not owner_id or not owner_id.strip():
+        raise HTTPException(status_code=422, detail="owner_id must be a non-empty string")
+    owner_id = owner_id.strip()
+
+    if file_ids is not None:
+        # Assign to specific files
+        updated = (
+            db.query(FileRecord)
+            .filter(FileRecord.id.in_(file_ids))
+            .update({FileRecord.owner_id: owner_id}, synchronize_session="fetch")
+        )
+    else:
+        # Assign to all currently unowned documents
+        updated = (
+            db.query(FileRecord)
+            .filter(FileRecord.owner_id.is_(None))
+            .update({FileRecord.owner_id: owner_id}, synchronize_session="fetch")
+        )
+
+    try:
+        db.commit()
+    except Exception as e:
+        db.rollback()
+        logger.exception(f"Error assigning owner: {e}")
+        raise HTTPException(status_code=500, detail="Failed to assign owner")
+
+    admin_name = get_current_owner_id(request) or "admin"
+    logger.info(f"Admin '{admin_name}' assigned owner_id='{owner_id}' to {updated} file(s)")
+    return {
+        "status": "success",
+        "message": f"Assigned owner to {updated} document(s)",
+        "updated_count": updated,
+        "owner_id": owner_id,
+    }

app/api/settings.py

@@ -446,6 +446,42 @@ async def install_ocr_languages(request: Request, admin: AdminUser):
         )
 
 
+@router.get("/{key}/suggestions")
+async def get_setting_suggestions(
+    key: str,
+    request: Request,
+    q: str = "",
+    limit: int = 10,
+):
+    """
+    Return autocomplete suggestions for a setting key.
+
+    Fetches values dynamically from cloud SDKs, installed tools, or
+    curated static lists depending on the setting.  Results are filtered
+    by case-insensitive substring match on the ``q`` parameter.
+
+    This endpoint does **not** require admin privileges so that the
+    autocomplete widget works for any authenticated user viewing settings.
+    """
+    from app.utils.suggestion_providers import SUGGESTION_PROVIDERS, get_suggestions  # noqa: PLC0415
+
+    if key not in SUGGESTION_PROVIDERS:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"No suggestions available for setting '{key}'",
+        )
+
+    try:
+        suggestions = get_suggestions(key, query=q, limit=max(1, min(limit, 50)))
+        return {"key": key, "suggestions": suggestions}
+    except Exception as e:
+        logger.error(f"Error fetching suggestions for {key}: {e}")
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="Failed to fetch suggestions",
+        )
+
+
 @router.get("/{key}/history")
 async def get_key_history(key: str, request: Request, db: DbSession, admin: AdminUser):
     """

app/api/user.py

@@ -4,14 +4,23 @@ User-related API endpoints
 
 import logging
 from hashlib import md5
+from typing import Annotated
 
-from fastapi import APIRouter, HTTPException, Request
+from fastapi import APIRouter, Depends, HTTPException, Query, Request
+from sqlalchemy import func
+from sqlalchemy.orm import Session
+
+from app.auth import require_login
+from app.database import get_db
+from app.models import FileRecord
 
 # Set up logging
 logger = logging.getLogger(__name__)
 
 router = APIRouter()
 
+DbSession = Annotated[Session, Depends(get_db)]
+
 
 async def whoami_handler(request: Request):
     """
@@ -46,3 +55,30 @@ async def whoami(request: Request):
 @router.get("/auth/whoami")
 async def auth_whoami(request: Request):
     return await whoami_handler(request)
+
+
+@router.get("/users/search")
+@require_login
+def search_known_users(
+    db: DbSession,
+    q: str = Query("", description="Substring to match against known owner IDs"),
+    limit: int = Query(5, ge=1, le=20, description="Maximum number of results"),
+):
+    """
+    Search known user identifiers (owner_ids) from existing documents.
+
+    Returns distinct ``owner_id`` values from the files table that contain
+    the query string as a case-insensitive substring.  Results are limited
+    to at most ``limit`` entries (default 5).
+
+    This powers the autocomplete widget on the settings page for the
+    ``default_owner_id`` field.
+    """
+    base_query = db.query(FileRecord.owner_id).filter(FileRecord.owner_id.isnot(None)).distinct()
+
+    if q.strip():
+        base_query = base_query.filter(func.lower(FileRecord.owner_id).contains(q.strip().lower()))
+
+    results = base_query.order_by(FileRecord.owner_id).limit(limit).all()
+
+    return {"users": [row[0] for row in results]}

app/config.py

@@ -116,6 +116,43 @@ class Settings(BaseSettings):
     session_secret: Optional[str] = None
     admin_group_name: str = "admin"
 
+    # Multi-user settings
+    multi_user_enabled: bool = Field(
+        default=False,
+        description=(
+            "Enable multi-user mode with individual document spaces per user. "
+            "When enabled, each authenticated user sees only their own documents, "
+            "uploads, and search results. Shared settings (AI, OCR) remain global. "
+            "Requires auth_enabled=True. Default: False (single-user/shared mode)."
+        ),
+    )
+    default_daily_upload_limit: int = Field(
+        default=0,
+        description=(
+            "Default maximum number of document uploads allowed per user per day "
+            "in multi-user mode. Set to 0 for unlimited. "
+            "Individual user limits can override this default. Default: 0 (unlimited)."
+        ),
+    )
+    unowned_docs_visible_to_all: bool = Field(
+        default=True,
+        description=(
+            "In multi-user mode, controls whether documents without an owner (owner_id is NULL) "
+            "are visible to all authenticated users. When True, unowned documents appear in every "
+            "user's file list alongside their own files. When False, only admins can see unowned "
+            "documents. Default: True."
+        ),
+    )
+    default_owner_id: Optional[str] = Field(
+        default=None,
+        description=(
+            "When set, automatically assigns this owner ID to newly ingested documents that would "
+            "otherwise have no owner (e.g. documents from IMAP, API without session, or legacy imports). "
+            "Use the admin /api/files/assign-owner endpoint to bulk-assign existing unclaimed documents. "
+            "Default: None (documents remain unowned until claimed)."
+        ),
+    )
+
     # Authentik
     authentik_client_id: Optional[str] = None
     authentik_client_secret: Optional[str] = None

app/models.py

@@ -24,6 +24,11 @@ class FileRecord(Base):
 
     id = Column(Integer, primary_key=True, index=True)
 
+    # Owner identifier for multi-user mode.
+    # Stores the user's unique identifier (e.g. email or OAuth sub claim).
+    # NULL means the file belongs to the shared/global space (single-user mode).
+    owner_id = Column(String, nullable=True, index=True)
+
     # Hash of the file content (e.g. SHA-256)
     # Note: duplicates are allowed so filehash is not unique
     filehash = Column(String, index=True, nullable=False)

app/tasks/convert_to_pdf.py

@@ -124,7 +124,9 @@ def _build_filename(file_path: str, original_filename: Optional[str], file_ext:
 
 
 @shared_task(bind=True)
-def convert_to_pdf(self, file_path: str, original_filename: Optional[str] = None) -> Optional[str]:
+def convert_to_pdf(
+    self, file_path: str, original_filename: Optional[str] = None, owner_id: Optional[str] = None
+) -> Optional[str]:
     """
     Converts a file to PDF using Gotenberg's API.
     Determines the appropriate Gotenberg endpoint based on the file's MIME type.
@@ -133,6 +135,7 @@ def convert_to_pdf(self, file_path: str, original_filename: Optional[str] = None
     Args:
         file_path: Path to the file to convert
         original_filename: Optional original filename (if different from path basename)
+        owner_id: Optional user identifier forwarded to process_document for multi-user mode.
     """
     task_id = self.request.id
     logger.info(f"[{task_id}] Starting PDF conversion: {file_path}")
@@ -332,9 +335,9 @@ def convert_to_pdf(self, file_path: str, original_filename: Optional[str] = None
                 # Change extension to .pdf for the original filename
                 original_base = os.path.splitext(original_filename)[0]
                 pdf_original_filename = f"{original_base}.pdf"
-                process_document.delay(converted_file_path, original_filename=pdf_original_filename)
+                process_document.delay(converted_file_path, original_filename=pdf_original_filename, owner_id=owner_id)
             else:
-                process_document.delay(converted_file_path)
+                process_document.delay(converted_file_path, owner_id=owner_id)
 
             return converted_file_path
         else:

app/tasks/process_document.py

@@ -25,7 +25,12 @@ logger = logging.getLogger(__name__)
 
 @celery.task(base=BaseTaskWithRetry, bind=True)
 def process_document(
-    self, original_local_file: str, original_filename: str = None, file_id: int = None, force_cloud_ocr: bool = False
+    self,
+    original_local_file: str,
+    original_filename: str = None,
+    file_id: int = None,
+    force_cloud_ocr: bool = False,
+    owner_id: str = None,
 ):
     """
     Process a document file and trigger appropriate text extraction.
@@ -37,6 +42,8 @@ def process_document(
                  detection and reuses the existing record (used for reprocessing).
         force_cloud_ocr: If True, forces Azure Document Intelligence OCR processing
                         regardless of embedded text quality. Used for re-processing.
+        owner_id: Optional user identifier for multi-user mode. When provided, the
+                  created FileRecord is associated with this user.
 
     Steps:
       1. Check if we have a FileRecord entry (via SHA-256 hash). If found, skip re-processing.
@@ -48,6 +55,10 @@ def process_document(
          - Otherwise, queue Azure Document Intelligence processing
       3. If force_cloud_ocr is True, skip local text extraction and use cloud OCR
     """
+    # Fall back to the configured default_owner_id when no explicit owner was provided
+    if owner_id is None and settings.default_owner_id:
+        owner_id = settings.default_owner_id
+
     task_id = self.request.id
     logger.info(f"[{task_id}] Starting document processing: {original_local_file}")
     log_task_progress(
@@ -141,6 +152,7 @@ def process_document(
                     mime_type=mime_type,
                     is_duplicate=True,
                     duplicate_of_id=existing.id,
+                    owner_id=owner_id,
                 )
                 db.add(duplicate_record)
                 db.commit()
@@ -191,6 +203,7 @@ def process_document(
                 file_size=file_size,
                 mime_type=mime_type,
                 is_duplicate=False,
+                owner_id=owner_id,
             )
             db.add(new_record)
             db.commit()

app/utils/settings_service.py

@@ -78,6 +78,52 @@ SETTING_METADATA = {
         "required": False,
         "restart_required": True,
     },
+    "multi_user_enabled": {
+        "category": "Authentication",
+        "description": (
+            "Enable multi-user mode with individual document spaces per user. "
+            "Each user sees only their own documents, uploads, and search results. "
+            "Requires auth_enabled=True."
+        ),
+        "type": "boolean",
+        "sensitive": False,
+        "required": False,
+        "restart_required": True,
+    },
+    "default_daily_upload_limit": {
+        "category": "Authentication",
+        "description": (
+            "Default maximum document uploads allowed per user per day in multi-user mode. Set to 0 for unlimited."
+        ),
+        "type": "integer",
+        "sensitive": False,
+        "required": False,
+        "restart_required": False,
+    },
+    "unowned_docs_visible_to_all": {
+        "category": "Authentication",
+        "description": (
+            "In multi-user mode, controls whether documents without an owner are visible to all users. "
+            "When True, unowned documents appear alongside each user's own files. "
+            "When False, only admins can see unowned documents."
+        ),
+        "type": "boolean",
+        "sensitive": False,
+        "required": False,
+        "restart_required": False,
+    },
+    "default_owner_id": {
+        "category": "Authentication",
+        "description": (
+            "Automatically assigns this owner ID to newly ingested documents that would otherwise "
+            "have no owner. Start typing to search existing users, or leave empty to keep documents "
+            "unowned until claimed."
+        ),
+        "type": "user_autocomplete",
+        "sensitive": False,
+        "required": False,
+        "restart_required": False,
+    },
     "session_secret": {
         "category": "Authentication",
         "description": "Secret key for session encryption (min 32 characters)",
@@ -314,7 +360,7 @@ SETTING_METADATA = {
     "azure_region": {
         "category": "AI Services",
         "description": "Azure region for Document Intelligence services (e.g., eastus)",
-        "type": "string",
+        "type": "autocomplete",
         "sensitive": False,
         "required": False,
         "restart_required": False,
@@ -366,8 +412,11 @@ SETTING_METADATA = {
     },
     "tesseract_language": {
         "category": "OCR Engines",
-        "description": "Tesseract language code(s), e.g. 'eng' or 'eng+deu'. Default: eng+deu (English + German).",
+        "description": (
-        "type": "string",
+            "Tesseract language code(s), e.g. 'eng' or 'eng+deu'. "
+            "Combine multiple with '+'. Default: eng+deu (English + German)."
+        ),
+        "type": "autocomplete",
         "sensitive": False,
         "required": False,
         "restart_required": False,
@@ -375,8 +424,8 @@ SETTING_METADATA = {
     # OCR – EasyOCR
     "easyocr_languages": {
         "category": "OCR Engines",
-        "description": "Comma-separated EasyOCR language codes, e.g. 'en,de,fr'. Default: en,de (English + German).",
+        "description": ("Comma-separated EasyOCR language codes, e.g. 'en,de,fr'. Default: en,de (English + German)."),
-        "type": "string",
+        "type": "autocomplete",
         "sensitive": False,
         "required": False,
         "restart_required": False,
@@ -807,7 +856,7 @@ SETTING_METADATA = {
     "aws_region": {
         "category": "Storage Providers",
         "description": "AWS region for S3 bucket (default: us-east-1)",
-        "type": "string",
+        "type": "autocomplete",
         "sensitive": False,
         "required": False,
         "restart_required": False,
@@ -1430,7 +1479,7 @@ SETTING_METADATA = {
             "Model name used for generating text embeddings via the OpenAI-compatible API. "
             "Embeddings drive the document similarity feature. Default: text-embedding-3-small."
         ),
-        "type": "string",
+        "type": "autocomplete",
         "sensitive": False,
         "required": False,
         "restart_required": False,

app/utils/suggestion_providers.py

@@ -0,0 +1,451 @@
+"""
+Dynamic suggestion providers for autocomplete-enabled settings.
+
+Each provider function returns a list of strings that are valid values
+for a particular setting.  Providers try to resolve values dynamically
+(e.g. by querying cloud SDKs or scanning installed software) and fall
+back to curated static lists when the runtime environment lacks the
+required libraries, credentials, or connectivity.
+
+**Fallback guarantee**: Every provider wraps its dynamic resolution in a
+``try/except Exception`` so that it *always* returns a usable list.
+Missing libraries (``ImportError``), missing credentials, network
+failures, or unexpected SDK errors all trigger a graceful fallback to
+the bundled static list.
+"""
+
+import logging
+import subprocess  # noqa: S404 — only used with fixed args, no user input
+from typing import Callable
+
+logger = logging.getLogger(__name__)
+
+# ---------------------------------------------------------------------------
+# AWS regions — fetched from boto3 if available
+# ---------------------------------------------------------------------------
+
+_AWS_REGIONS_STATIC: list[str] = [
+    "af-south-1",
+    "ap-east-1",
+    "ap-northeast-1",
+    "ap-northeast-2",
+    "ap-northeast-3",
+    "ap-south-1",
+    "ap-south-2",
+    "ap-southeast-1",
+    "ap-southeast-2",
+    "ap-southeast-3",
+    "ap-southeast-4",
+    "ca-central-1",
+    "ca-west-1",
+    "eu-central-1",
+    "eu-central-2",
+    "eu-north-1",
+    "eu-south-1",
+    "eu-south-2",
+    "eu-west-1",
+    "eu-west-2",
+    "eu-west-3",
+    "il-central-1",
+    "me-central-1",
+    "me-south-1",
+    "sa-east-1",
+    "us-east-1",
+    "us-east-2",
+    "us-west-1",
+    "us-west-2",
+]
+
+
+def get_aws_regions() -> list[str]:
+    """Return available AWS S3 regions via boto3, falling back to a static list."""
+    try:
+        import boto3  # noqa: PLC0415
+
+        session = boto3.session.Session()
+        regions = sorted(session.get_available_regions("s3"))
+        if regions:
+            return regions
+    except Exception:
+        logger.debug("boto3 not available or failed; using static AWS region list")
+    return _AWS_REGIONS_STATIC
+
+
+# ---------------------------------------------------------------------------
+# Azure regions — resolved from known Cognitive Services locations
+# ---------------------------------------------------------------------------
+
+_AZURE_REGIONS_STATIC: list[str] = [
+    "australiacentral",
+    "australiaeast",
+    "australiasoutheast",
+    "brazilsouth",
+    "canadacentral",
+    "canadaeast",
+    "centralindia",
+    "centralus",
+    "eastasia",
+    "eastus",
+    "eastus2",
+    "francecentral",
+    "germanywestcentral",
+    "japaneast",
+    "japanwest",
+    "koreacentral",
+    "koreasouth",
+    "northcentralus",
+    "northeurope",
+    "norwayeast",
+    "polandcentral",
+    "qatarcentral",
+    "southafricanorth",
+    "southcentralus",
+    "southeastasia",
+    "swedencentral",
+    "switzerlandnorth",
+    "uaenorth",
+    "uksouth",
+    "ukwest",
+    "westcentralus",
+    "westeurope",
+    "westus",
+    "westus2",
+    "westus3",
+]
+
+
+def get_azure_regions() -> list[str]:
+    """Return Azure Cognitive Services regions.
+
+    Falls back to a curated static list because there is no
+    unauthenticated public endpoint to enumerate regions.
+    """
+    return _AZURE_REGIONS_STATIC
+
+
+# ---------------------------------------------------------------------------
+# Tesseract languages — probed from `tesseract --list-langs`
+# ---------------------------------------------------------------------------
+
+_TESSERACT_LANGS_STATIC: list[str] = [
+    "afr",
+    "amh",
+    "ara",
+    "asm",
+    "aze",
+    "bel",
+    "ben",
+    "bod",
+    "bos",
+    "bre",
+    "bul",
+    "cat",
+    "ceb",
+    "ces",
+    "chi_sim",
+    "chi_tra",
+    "chr",
+    "cos",
+    "cym",
+    "dan",
+    "deu",
+    "div",
+    "ell",
+    "eng",
+    "enm",
+    "epo",
+    "est",
+    "eus",
+    "fao",
+    "fas",
+    "fil",
+    "fin",
+    "fra",
+    "frk",
+    "frm",
+    "fry",
+    "gla",
+    "gle",
+    "glg",
+    "grc",
+    "guj",
+    "hat",
+    "heb",
+    "hin",
+    "hrv",
+    "hun",
+    "hye",
+    "iku",
+    "ind",
+    "isl",
+    "ita",
+    "jav",
+    "jpn",
+    "kan",
+    "kat",
+    "kaz",
+    "khm",
+    "kir",
+    "kor",
+    "lao",
+    "lat",
+    "lav",
+    "lit",
+    "ltz",
+    "mal",
+    "mar",
+    "mkd",
+    "mlt",
+    "mon",
+    "mri",
+    "msa",
+    "mya",
+    "nep",
+    "nld",
+    "nor",
+    "oci",
+    "ori",
+    "pan",
+    "pol",
+    "por",
+    "pus",
+    "que",
+    "ron",
+    "rus",
+    "san",
+    "sin",
+    "slk",
+    "slv",
+    "snd",
+    "spa",
+    "sqi",
+    "srp",
+    "sun",
+    "swa",
+    "swe",
+    "syr",
+    "tam",
+    "tat",
+    "tel",
+    "tgk",
+    "tha",
+    "tir",
+    "ton",
+    "tur",
+    "uig",
+    "ukr",
+    "urd",
+    "uzb",
+    "vie",
+    "yid",
+    "yor",
+]
+
+
+def get_tesseract_languages() -> list[str]:
+    """Return installed Tesseract language codes, falling back to a static list."""
+    try:
+        result = subprocess.run(  # noqa: S603, S607
+            ["tesseract", "--list-langs"],  # noqa: S607
+            capture_output=True,
+            text=True,
+            timeout=5,
+            check=False,
+        )
+        if result.returncode == 0:
+            lines = result.stdout.strip().splitlines()
+            # First line is the header ("List of available languages ...")
+            langs = sorted(line.strip() for line in lines[1:] if line.strip())
+            if langs:
+                return langs
+    except Exception:
+        logger.debug("tesseract not available; using static language list")
+    return _TESSERACT_LANGS_STATIC
+
+
+# ---------------------------------------------------------------------------
+# EasyOCR languages — probed from the easyocr module
+# ---------------------------------------------------------------------------
+
+_EASYOCR_LANGS_STATIC: list[str] = [
+    "abq",
+    "ady",
+    "af",
+    "ang",
+    "ar",
+    "as",
+    "ava",
+    "az",
+    "be",
+    "bg",
+    "bh",
+    "bn",
+    "bs",
+    "ch_sim",
+    "ch_tra",
+    "che",
+    "cs",
+    "cy",
+    "da",
+    "dar",
+    "de",
+    "en",
+    "es",
+    "et",
+    "fa",
+    "fi",
+    "fr",
+    "ga",
+    "gom",
+    "hi",
+    "hr",
+    "hu",
+    "id",
+    "inh",
+    "is",
+    "it",
+    "ja",
+    "ka",
+    "kk",
+    "km",
+    "kn",
+    "ko",
+    "ku",
+    "la",
+    "lbe",
+    "lez",
+    "lt",
+    "lv",
+    "mah",
+    "mai",
+    "mi",
+    "mn",
+    "mr",
+    "ms",
+    "mt",
+    "ne",
+    "new",
+    "nl",
+    "no",
+    "oc",
+    "pi",
+    "pl",
+    "pt",
+    "ro",
+    "ru",
+    "rs_cyrillic",
+    "rs_latin",
+    "sa",
+    "sck",
+    "sk",
+    "sl",
+    "sq",
+    "sv",
+    "sw",
+    "ta",
+    "tab",
+    "te",
+    "th",
+    "tjk",
+    "tl",
+    "tr",
+    "ug",
+    "uk",
+    "ur",
+    "uz",
+    "vi",
+]
+
+
+def get_easyocr_languages() -> list[str]:
+    """Return supported EasyOCR language codes, falling back to a static list."""
+    try:
+        import easyocr  # noqa: PLC0415
+
+        # easyocr stores the language list internally
+        if hasattr(easyocr, "config") and hasattr(easyocr.config, "all_lang_list"):
+            return sorted(easyocr.config.all_lang_list)
+    except Exception:
+        logger.debug("easyocr not available; using static language list")
+    return _EASYOCR_LANGS_STATIC
+
+
+# ---------------------------------------------------------------------------
+# Embedding models — static list (no standard discovery API)
+# ---------------------------------------------------------------------------
+
+_EMBEDDING_MODELS: list[str] = [
+    "text-embedding-3-small",
+    "text-embedding-3-large",
+    "text-embedding-ada-002",
+    "nomic-embed-text",
+    "nomic-embed-text-v1.5",
+    "mxbai-embed-large",
+    "mxbai-embed-large-v1",
+    "all-MiniLM-L6-v2",
+    "all-MiniLM-L12-v2",
+    "bge-small-en-v1.5",
+    "bge-base-en-v1.5",
+    "bge-large-en-v1.5",
+    "e5-small-v2",
+    "e5-base-v2",
+    "e5-large-v2",
+    "gte-small",
+    "gte-base",
+    "gte-large",
+    "voyage-3",
+    "voyage-3-lite",
+    "voyage-code-3",
+]
+
+
+def get_embedding_models() -> list[str]:
+    """Return known embedding model names."""
+    return _EMBEDDING_MODELS
+
+
+# ---------------------------------------------------------------------------
+# Registry — maps setting keys to their provider functions
+# ---------------------------------------------------------------------------
+
+SUGGESTION_PROVIDERS: dict[str, Callable[[], list[str]]] = {
+    "aws_region": get_aws_regions,
+    "azure_region": get_azure_regions,
+    "tesseract_language": get_tesseract_languages,
+    "easyocr_languages": get_easyocr_languages,
+    "embedding_model": get_embedding_models,
+}
+
+
+def get_suggestions(key: str, query: str = "", limit: int = 10) -> list[str]:
+    """
+    Return autocomplete suggestions for the given setting key.
+
+    Fetches the full list from the registered provider, filters by
+    case-insensitive substring match on *query*, and returns at most
+    *limit* results.
+
+    Args:
+        key: The setting key (must be registered in SUGGESTION_PROVIDERS).
+        query: Substring to filter by (case-insensitive).
+        limit: Maximum number of results to return.
+
+    Returns:
+        Filtered list of suggestion strings.
+
+    Raises:
+        KeyError: If no provider is registered for *key*.
+    """
+    provider = SUGGESTION_PROVIDERS.get(key)
+    if provider is None:
+        raise KeyError(f"No suggestion provider registered for setting '{key}'")
+
+    all_values = provider()
+    q = query.strip().lower()
+
+    if q:
+        filtered = [v for v in all_values if q in v.lower()]
+    else:
+        filtered = list(all_values)
+
+    return filtered[:limit]

app/utils/user_scope.py

@@ -0,0 +1,84 @@
+"""
+User-scoping utilities for multi-user document isolation.
+
+When ``multi_user_enabled`` is ``True`` in settings, every document query
+is filtered by the authenticated user's identifier so that each user sees
+only their own documents.  When the flag is ``False`` (default), all
+documents are visible to all users (single-user / shared mode).
+"""
+
+import logging
+
+from fastapi import Request
+from sqlalchemy import or_
+from sqlalchemy.orm import Query
+from sqlalchemy.sql import false
+
+from app.config import settings
+from app.models import FileRecord
+
+logger = logging.getLogger(__name__)
+
+
+def get_current_owner_id(request: Request) -> str | None:
+    """Extract the owner identifier for the current authenticated user.
+
+    The owner ID is derived from the user's session data.  It uses the
+    ``sub`` claim (OAuth subject) when available, falling back to
+    ``preferred_username`` or ``email``.  Returns ``None`` when no user
+    is authenticated.
+
+    Args:
+        request: The current FastAPI request with session data.
+
+    Returns:
+        A stable string identifier for the user, or ``None``.
+    """
+    user = request.session.get("user")
+    if not user or not isinstance(user, dict):
+        return None
+    # Prefer 'sub' (OAuth subject), then 'preferred_username', then 'email', then 'id'
+    return user.get("sub") or user.get("preferred_username") or user.get("email") or user.get("id")
+
+
+def apply_owner_filter(query: Query, request: Request) -> Query:
+    """Conditionally filter a ``FileRecord`` query by the current user.
+
+    When multi-user mode is enabled, only files whose ``owner_id``
+    matches the authenticated user are returned.  Admin users bypass
+    the filter and see all documents.
+
+    When ``unowned_docs_visible_to_all`` is ``True`` (default), documents
+    with ``owner_id IS NULL`` (unclaimed) are also included for every
+    authenticated user so they can be discovered and claimed.
+
+    When multi-user mode is disabled the query is returned unchanged.
+
+    Args:
+        query: A SQLAlchemy query selecting ``FileRecord`` rows.
+        request: The current FastAPI request (for session inspection).
+
+    Returns:
+        The (possibly filtered) query.
+    """
+    if not settings.multi_user_enabled:
+        return query
+
+    user = request.session.get("user")
+    if isinstance(user, dict) and user.get("is_admin"):
+        # Admins see all documents in multi-user mode
+        return query
+
+    owner_id = get_current_owner_id(request)
+    if owner_id is None:
+        # No authenticated user — return empty result set
+        return query.filter(false())
+
+    # Build filter: user's own documents
+    conditions = [FileRecord.owner_id == owner_id]
+
+    # Optionally include unclaimed (owner_id IS NULL) documents
+    if settings.unowned_docs_visible_to_all:
+        conditions.append(FileRecord.owner_id.is_(None))
+
+    return query.filter(or_(*conditions))

docs/API.md

@@ -614,6 +614,150 @@ curl -X POST "http://<your-instance>/api/files/bulk-download" \
 **Error Responses**:
 - `404`: No files found with the provided IDs, or none of the selected files exist on disk
 
+### Document Ownership (Multi-User Mode)
+
+These endpoints are available when `MULTI_USER_ENABLED=true`.
+
+---
+
+**POST** `/api/files/{file_id}/claim`
+
+Claim an unclaimed document (owner_id is NULL) for the current user.
+
+```bash
+curl -X POST "http://<your-instance>/api/files/42/claim"
+```
+
+**Response**:
+```json
+{
+  "status": "success",
+  "message": "Document claimed successfully",
+  "file_id": 42,
+  "owner_id": "alice@example.com"
+}
+```
+
+**Error Responses**:
+- `400`: Multi-user mode is not enabled
+- `401`: Authentication required
+- `403`: Document is already owned by another user
+
+---
+
+**POST** `/api/files/bulk-claim`
+
+Claim multiple unclaimed documents at once. Already-owned documents are skipped.
+
+**Request body**: JSON array of file IDs
+
+```bash
+curl -X POST "http://<your-instance>/api/files/bulk-claim" \
+  -H "Content-Type: application/json" \
+  -d '[1, 2, 3]'
+```
+
+**Response**:
+```json
+{
+  "status": "success",
+  "claimed_count": 2,
+  "claimed_ids": [1, 3],
+  "skipped": [{"file_id": 2, "reason": "already owned"}],
+  "owner_id": "alice@example.com"
+}
+```
+
+---
+
+**POST** `/api/files/assign-owner`
+
+**Admin only.** Assign an owner to documents. If `file_ids` body is omitted, assigns all
+currently unclaimed documents to the specified owner.
+
+**Query Parameters**:
+- `owner_id` (required): The user identifier to assign
+
+**Request body** (optional): JSON array of specific file IDs
+
+```bash
+# Assign all unclaimed documents to a user
+curl -X POST "http://<your-instance>/api/files/assign-owner?owner_id=alice@example.com"
+
+# Assign specific files
+curl -X POST "http://<your-instance>/api/files/assign-owner?owner_id=alice@example.com" \
+  -H "Content-Type: application/json" \
+  -d '[1, 2, 3]'
+```
+
+**Response**:
+```json
+{
+  "status": "success",
+  "message": "Assigned owner to 5 document(s)",
+  "updated_count": 5,
+  "owner_id": "alice@example.com"
+}
+```
+
+**Error Responses**:
+- `400`: Multi-user mode is not enabled
+- `403`: Only admins can assign document owners
+
+---
+
+**GET** `/api/users/search`
+
+Search known user identifiers from existing documents. Powers the autocomplete widget
+in the settings page for the `DEFAULT_OWNER_ID` field.
+
+**Query Parameters**:
+- `q` (optional): Substring to match against known owner IDs (case-insensitive)
+- `limit` (optional): Maximum results to return (default: 5, max: 20)
+
+```bash
+curl "http://<your-instance>/api/users/search?q=risti&limit=5"
+```
+
+**Response**:
+```json
+{
+  "users": ["christianlouis"]
+}
+```
+
+---
+
+### Settings Suggestions (Autocomplete)
+
+**GET** `/api/settings/{key}/suggestions`
+
+Return dynamic autocomplete suggestions for a setting. Providers attempt to
+resolve values from cloud SDKs or installed tools and fall back to curated
+static lists when unavailable.
+
+**Supported keys**: `aws_region`, `azure_region`, `tesseract_language`,
+`easyocr_languages`, `embedding_model`
+
+**Query Parameters**:
+- `q` (optional): Substring to filter suggestions (case-insensitive)
+- `limit` (optional): Maximum results to return (default: 10, max: 50)
+
+```bash
+curl "http://<your-instance>/api/settings/aws_region/suggestions?q=east&limit=5"
+```
+
+**Response**:
+```json
+{
+  "key": "aws_region",
+  "suggestions": ["ap-east-1", "ap-northeast-1", "ap-southeast-1", "us-east-1", "us-east-2"]
+}
+```
+
+**Error Responses**:
+- `404`: No suggestion provider registered for the given key
+
 ### File Preview
 
 **GET** `/api/files/{file_id}/preview`

docs/ConfigurationGuide.md

@@ -149,6 +149,52 @@ DocuElevate can monitor multiple IMAP mailboxes for document attachments. Each m
 | `AUTHENTIK_CONFIG_URL`  | Configuration URL for Authentik OpenID Connect.             |
 | `OAUTH_PROVIDER_NAME`   | Display name for the OAuth provider button.                  |
 
+### Multi-User Mode
+
+When multi-user mode is enabled, each authenticated user gets their own isolated document space.
+Uploads, search results, and file management are scoped to the individual user. Shared settings
+(AI configuration, OCR providers, storage destinations) remain global.
+
+Admin users (determined by `ADMIN_GROUP_NAME`) bypass the user filter and can see all documents.
+
+Requires `AUTH_ENABLED=true`.
+
+| **Variable**                | **Description**                                                                 | **Default** |
+|-----------------------------|---------------------------------------------------------------------------------|-------------|
+| `MULTI_USER_ENABLED`        | Enable multi-user mode with individual document spaces per user.               | `false`     |
+| `DEFAULT_DAILY_UPLOAD_LIMIT`| Maximum document uploads allowed per user per day. `0` = unlimited.            | `0`         |
+| `UNOWNED_DOCS_VISIBLE_TO_ALL` | Show unclaimed documents (no owner) to all users. When `false`, only admins see them. | `true` |
+| `DEFAULT_OWNER_ID`          | Automatically assign this owner to newly ingested documents without a session (e.g. IMAP, API). Leave empty to keep unowned. | *(empty)* |
+
+#### Unclaimed Documents
+
+Documents ingested without a user session (e.g. via IMAP polling, API calls without authentication,
+or legacy imports) have `owner_id = NULL`. These are called **unclaimed** documents.
+
+- When `UNOWNED_DOCS_VISIBLE_TO_ALL=true` (default), every authenticated user sees unclaimed
+  documents alongside their own files. This allows users to discover and claim them.
+- When `UNOWNED_DOCS_VISIBLE_TO_ALL=false`, only admins can see unclaimed documents.
+
+#### Claiming Documents
+
+Users can claim unclaimed documents via the API:
+
+- **`POST /api/files/{file_id}/claim`** — Claim a single unclaimed document.
+- **`POST /api/files/bulk-claim`** — Claim multiple unclaimed documents at once.
+
+Only documents with `owner_id = NULL` can be claimed. Already-owned documents cannot be claimed
+by another user.
+
+#### Admin Owner Assignment
+
+Admins can assign ownership of documents to any user:
+
+- **`POST /api/files/assign-owner?owner_id=<user_id>`** — Assign all unclaimed documents to
+  the specified user, or pass a `file_ids` JSON body to assign specific files.
+
+The `DEFAULT_OWNER_ID` setting can also be configured via the Settings page, which provides an
+autocomplete field that searches existing users by substring.
+
 ### Security Headers
 
 DocuElevate supports HTTP security headers to improve browser-side security. **These headers are disabled by default** since most deployments use a reverse proxy (Traefik, Nginx, etc.) that already adds them. Enable only if deploying directly without a reverse proxy. See [Deployment Guide - Security Headers](DeploymentGuide.md#security-headers) for detailed configuration examples.
@@ -1059,6 +1105,10 @@ AUTHENTIK_CLIENT_SECRET=...
 AUTHENTIK_CONFIG_URL=https://auth.example.com/.well-known/openid-configuration
 OAUTH_PROVIDER_NAME=Authentik SSO
 
+# Multi-user mode (requires AUTH_ENABLED=true)
+MULTI_USER_ENABLED=false
+DEFAULT_DAILY_UPLOAD_LIMIT=0
+
 # Storage services
 PAPERLESS_NGX_API_TOKEN=...
 PAPERLESS_HOST=https://paperless.example.com

docs/SettingsManagement.md

@@ -48,6 +48,8 @@ Settings are organized into logical categories for easy navigation:
 - **Dropdown**: Predefined option lists (e.g., PDF/A format, S3 storage class, S3 ACL)
 - **Multi-select**: Comma-separated selections from a list (e.g., OCR providers)
 - **Model Picker**: Free-text with suggested model names (e.g., AI model selection)
+- **User Autocomplete**: Typeahead search for existing user identifiers (e.g., default owner assignment), fetches from `GET /api/users/search`
+- **Autocomplete**: Typeahead search with dynamic suggestions fetched from `GET /api/settings/{key}/suggestions`. Used for AWS/Azure regions, OCR language codes, and embedding models. Providers attempt dynamic resolution (e.g., boto3 for AWS regions, `tesseract --list-langs` for Tesseract) and fall back to curated static lists if the SDK or tool is unavailable.
 - **List**: Comma-separated values (notification URLs, CORS origins)
 
 ### Sensitive Data

frontend/templates/settings.html

@@ -290,6 +290,155 @@
                       {% endfor %}
                     </div>
                     <p class="mt-1 text-xs text-gray-400">Effective value: <code x-text="formData['{{ setting.key }}'] || '(none)'"></code></p>
+                    {% elif setting.metadata.type == 'user_autocomplete' %}
+                    <!-- User Autocomplete (WCAG 2.1 AA combobox pattern) -->
+                    <div class="relative" x-data="userAutocomplete('{{ setting.key }}')" @click.away="showSuggestions = false">
+                      <div class="relative">
+                        <label for="{{ setting.key }}" class="sr-only">{{ setting.key.replace('_', ' ').title() }}</label>
+                        <input
+                          type="text"
+                          id="{{ setting.key }}"
+                          name="{{ setting.key }}"
+                          x-model="formData['{{ setting.key }}']"
+                          @input.debounce.250ms="fetchSuggestions($event.target.value)"
+                          @focus="if (formData['{{ setting.key }}']) fetchSuggestions(formData['{{ setting.key }}']); else fetchSuggestions('')"
+                          @keydown.arrow-down.prevent="highlightNext()"
+                          @keydown.arrow-up.prevent="highlightPrev()"
+                          @keydown.enter.prevent="selectHighlighted()"
+                          @keydown.escape="showSuggestions = false"
+                          class="setting-input w-full px-3 py-2 pl-9 border border-gray-300 rounded-md shadow-sm focus:outline-none focus:ring-blue-500 focus:border-blue-500"
+                          placeholder="Start typing to search users…"
+                          autocomplete="off"
+                          role="combobox"
+                          aria-autocomplete="list"
+                          aria-label="{{ setting.key.replace('_', ' ').title() }} — search existing users"
+                          :aria-expanded="showSuggestions && suggestions.length > 0"
+                          aria-haspopup="listbox"
+                          :aria-controls="'{{ setting.key }}_listbox'"
+                          :aria-activedescendant="highlightedIdx >= 0 ? '{{ setting.key }}_opt_' + highlightedIdx : ''"
+                        />
+                        <div class="absolute inset-y-0 left-0 flex items-center pl-3 pointer-events-none">
+                          <i class="fas fa-user text-gray-400 text-sm" aria-hidden="true"></i>
+                        </div>
+                        <div x-show="loading" class="absolute inset-y-0 right-0 flex items-center pr-3" aria-hidden="true">
+                          <i class="fas fa-spinner fa-spin text-gray-400 text-sm"></i>
+                        </div>
+                      </div>
+                      <!-- Live region for screen reader announcements -->
+                      <div class="sr-only" aria-live="polite" aria-atomic="true" x-text="statusText"></div>
+                      <!-- Suggestion dropdown -->
+                      <ul
+                        x-show="showSuggestions && suggestions.length > 0"
+                        x-transition:enter="transition ease-out duration-100"
+                        x-transition:enter-start="opacity-0 -translate-y-1"
+                        x-transition:enter-end="opacity-100 translate-y-0"
+                        class="absolute z-50 w-full mt-1 bg-white border border-gray-200 rounded-md shadow-lg max-h-48 overflow-y-auto"
+                        role="listbox"
+                        :id="'{{ setting.key }}_listbox'"
+                        :aria-label="'{{ setting.key.replace('_', ' ').title() }} suggestions'"
+                      >
+                        <template x-for="(user, idx) in suggestions" :key="user">
+                          <li
+                            :id="'{{ setting.key }}_opt_' + idx"
+                            role="option"
+                            :aria-selected="highlightedIdx === idx"
+                            @mouseenter="highlightedIdx = idx"
+                            @click="selectUser(user)"
+                            class="px-3 py-2.5 text-sm cursor-pointer flex items-center gap-2 min-h-[44px]"
+                            :class="highlightedIdx === idx ? 'bg-blue-50 text-blue-700' : 'text-gray-700 hover:bg-gray-50'"
+                          >
+                            <i class="fas fa-user-circle text-gray-400" aria-hidden="true"></i>
+                            <span x-text="user"></span>
+                          </li>
+                        </template>
+                      </ul>
+                      <!-- Empty state -->
+                      <div
+                        x-show="showSuggestions && suggestions.length === 0 && !loading && searchDone"
+                        class="absolute z-50 w-full mt-1 bg-white border border-gray-200 rounded-md shadow-lg px-3 py-2 text-sm text-gray-500"
+                        role="status"
+                      >
+                        <i class="fas fa-info-circle mr-1" aria-hidden="true"></i> No matching users found
+                      </div>
+                      <p class="text-xs text-gray-400 mt-1" id="{{ setting.key }}_help">
+                        <i class="fas fa-info-circle mr-1" aria-hidden="true"></i>
+                        Type to search existing users by name, or enter any identifier manually.
+                      </p>
+                    </div>
+                    {% elif setting.metadata.type == 'autocomplete' %}
+                    <!-- Dynamic Autocomplete (WCAG 2.1 AA combobox pattern) -->
+                    <div class="relative" x-data="dynamicAutocomplete('{{ setting.key }}')" @click.away="showSuggestions = false">
+                      <div class="relative">
+                        <label for="{{ setting.key }}" class="sr-only">{{ setting.key.replace('_', ' ').title() }}</label>
+                        <input
+                          type="text"
+                          id="{{ setting.key }}"
+                          name="{{ setting.key }}"
+                          x-model="formData['{{ setting.key }}']"
+                          @input.debounce.200ms="fetchSuggestions($event.target.value)"
+                          @focus="fetchSuggestions(formData['{{ setting.key }}'] || '')"
+                          @keydown.arrow-down.prevent="highlightNext()"
+                          @keydown.arrow-up.prevent="highlightPrev()"
+                          @keydown.enter.prevent="selectHighlighted()"
+                          @keydown.escape="showSuggestions = false"
+                          class="setting-input w-full px-3 py-2 pl-9 border border-gray-300 rounded-md shadow-sm focus:outline-none focus:ring-blue-500 focus:border-blue-500"
+                          placeholder="Type to search or enter a value…"
+                          autocomplete="off"
+                          role="combobox"
+                          aria-autocomplete="list"
+                          aria-label="{{ setting.key.replace('_', ' ').title() }} — type to search suggestions"
+                          :aria-expanded="showSuggestions && filtered.length > 0"
+                          aria-haspopup="listbox"
+                          :aria-controls="'{{ setting.key }}_listbox'"
+                          :aria-activedescendant="highlightedIdx >= 0 ? '{{ setting.key }}_opt_' + highlightedIdx : ''"
+                        />
+                        <div class="absolute inset-y-0 left-0 flex items-center pl-3 pointer-events-none">
+                          <i class="fas fa-search text-gray-400 text-sm" aria-hidden="true"></i>
+                        </div>
+                        <div x-show="loading" class="absolute inset-y-0 right-0 flex items-center pr-3" aria-hidden="true">
+                          <i class="fas fa-spinner fa-spin text-gray-400 text-sm"></i>
+                        </div>
+                      </div>
+                      <!-- Live region for screen reader announcements -->
+                      <div class="sr-only" aria-live="polite" aria-atomic="true" x-text="statusText"></div>
+                      <!-- Suggestion dropdown -->
+                      <ul
+                        x-show="showSuggestions && filtered.length > 0"
+                        x-transition:enter="transition ease-out duration-100"
+                        x-transition:enter-start="opacity-0 -translate-y-1"
+                        x-transition:enter-end="opacity-100 translate-y-0"
+                        class="absolute z-50 w-full mt-1 bg-white border border-gray-200 rounded-md shadow-lg max-h-48 overflow-y-auto"
+                        role="listbox"
+                        :id="'{{ setting.key }}_listbox'"
+                        :aria-label="'{{ setting.key.replace('_', ' ').title() }} suggestions'"
+                      >
+                        <template x-for="(item, idx) in filtered" :key="item">
+                          <li
+                            :id="'{{ setting.key }}_opt_' + idx"
+                            role="option"
+                            :aria-selected="highlightedIdx === idx"
+                            @mouseenter="highlightedIdx = idx"
+                            @click="selectItem(item)"
+                            class="px-3 py-2.5 text-sm cursor-pointer min-h-[44px] flex items-center"
+                            :class="highlightedIdx === idx ? 'bg-blue-50 text-blue-700' : 'text-gray-700 hover:bg-gray-50'"
+                          >
+                            <span x-text="item"></span>
+                          </li>
+                        </template>
+                      </ul>
+                      <!-- Empty state -->
+                      <div
+                        x-show="showSuggestions && filtered.length === 0 && !loading && searchDone"
+                        class="absolute z-50 w-full mt-1 bg-white border border-gray-200 rounded-md shadow-lg px-3 py-2 text-sm text-gray-500"
+                        role="status"
+                      >
+                        <i class="fas fa-info-circle mr-1" aria-hidden="true"></i> No matches — you can still type a custom value
+                      </div>
+                      <p class="text-xs text-gray-400 mt-1">
+                        <i class="fas fa-info-circle mr-1" aria-hidden="true"></i>
+                        Type to search known values, or enter any custom value.
+                      </p>
+                    </div>
                     {% elif setting.metadata.type == 'model_picker' %}
                     <!-- Model Picker -->
                     <div class="relative">
@@ -712,5 +861,147 @@ function settingsApp() {
     }
   };
 }
+
+/**
+ * Alpine.js component for the user-autocomplete setting widget.
+ * Implements the ARIA 1.1 combobox pattern (WCAG 2.1 AA).
+ * Fetches matching owner_id values from GET /api/users/search?q=<term>&limit=5
+ * and displays them in a dropdown list.
+ */
+function userAutocomplete(settingKey) {
+  return {
+    suggestions: [],
+    showSuggestions: false,
+    loading: false,
+    searchDone: false,
+    highlightedIdx: -1,
+    statusText: '',
+
+    async fetchSuggestions(query) {
+      this.loading = true;
+      this.searchDone = false;
+      this.highlightedIdx = -1;
+      this.statusText = 'Searching…';
+      try {
+        const resp = await fetch('/api/users/search?q=' + encodeURIComponent(query || '') + '&limit=5');
+        if (resp.ok) {
+          const data = await resp.json();
+          this.suggestions = data.users || [];
+        } else {
+          this.suggestions = [];
+        }
+      } catch {
+        this.suggestions = [];
+      }
+      this.loading = false;
+      this.searchDone = true;
+      this.showSuggestions = true;
+      // Announce result count to screen readers via live region
+      if (this.suggestions.length === 0) {
+        this.statusText = 'No matching users found';
+      } else if (this.suggestions.length === 1) {
+        this.statusText = '1 user found';
+      } else {
+        this.statusText = this.suggestions.length + ' users found';
+      }
+    },
+
+    selectUser(user) {
+      this.formData[settingKey] = user;
+      this.showSuggestions = false;
+      this.statusText = user + ' selected';
+    },
+
+    highlightNext() {
+      if (this.suggestions.length === 0) return;
+      this.highlightedIdx = (this.highlightedIdx + 1) % this.suggestions.length;
+      this.statusText = this.suggestions[this.highlightedIdx];
+    },
+
+    highlightPrev() {
+      if (this.suggestions.length === 0) return;
+      this.highlightedIdx = this.highlightedIdx <= 0 ? this.suggestions.length - 1 : this.highlightedIdx - 1;
+      this.statusText = this.suggestions[this.highlightedIdx];
+    },
+
+    selectHighlighted() {
+      if (this.highlightedIdx >= 0 && this.highlightedIdx < this.suggestions.length) {
+        this.selectUser(this.suggestions[this.highlightedIdx]);
+      } else {
+        this.showSuggestions = false;
+      }
+    }
+  };
+}
+
+/**
+ * Alpine.js component for the generic dynamic autocomplete setting widget.
+ * Fetches suggestions from GET /api/settings/{key}/suggestions?q=<term>&limit=10
+ * using a debounced API call.  Implements the ARIA 1.1 combobox pattern.
+ */
+function dynamicAutocomplete(settingKey) {
+  return {
+    filtered: [],
+    showSuggestions: false,
+    loading: false,
+    searchDone: false,
+    highlightedIdx: -1,
+    statusText: '',
+
+    async fetchSuggestions(query) {
+      this.loading = true;
+      this.searchDone = false;
+      this.highlightedIdx = -1;
+      this.statusText = 'Searching…';
+      try {
+        const resp = await fetch('/api/settings/' + encodeURIComponent(settingKey) + '/suggestions?q=' + encodeURIComponent(query || '') + '&limit=10');
+        if (resp.ok) {
+          const data = await resp.json();
+          this.filtered = data.suggestions || [];
+        } else {
+          this.filtered = [];
+        }
+      } catch {
+        this.filtered = [];
+      }
+      this.loading = false;
+      this.searchDone = true;
+      this.showSuggestions = true;
+      if (this.filtered.length === 0) {
+        this.statusText = 'No matches found';
+      } else if (this.filtered.length === 1) {
+        this.statusText = '1 suggestion found';
+      } else {
+        this.statusText = this.filtered.length + ' suggestions found';
+      }
+    },
+
+    selectItem(item) {
+      this.formData[settingKey] = item;
+      this.showSuggestions = false;
+      this.statusText = item + ' selected';
+    },
+
+    highlightNext() {
+      if (this.filtered.length === 0) return;
+      this.highlightedIdx = (this.highlightedIdx + 1) % this.filtered.length;
+      this.statusText = this.filtered[this.highlightedIdx];
+    },
+
+    highlightPrev() {
+      if (this.filtered.length === 0) return;
+      this.highlightedIdx = this.highlightedIdx <= 0 ? this.filtered.length - 1 : this.highlightedIdx - 1;
+      this.statusText = this.filtered[this.highlightedIdx];
+    },
+
+    selectHighlighted() {
+      if (this.highlightedIdx >= 0 && this.highlightedIdx < this.filtered.length) {
+        this.selectItem(this.filtered[this.highlightedIdx]);
+      } else {
+        this.showSuggestions = false;
+      }
+    }
+  };
+}
 </script>
 {% endblock %}

migrations/versions/012_add_multi_user_support.py

@@ -0,0 +1,29 @@
+"""Add multi-user support: owner_id column on files table
+
+Revision ID: 012_add_multi_user_support
+Revises: 011_add_pdfa_paths
+Create Date: 2026-03-05
+
+"""
+
+from typing import Union
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "012_add_multi_user_support"
+down_revision: Union[str, None] = "011_add_pdfa_paths"
+depends_on: Union[str, None] = None
+
+
+def upgrade() -> None:
+    """Add owner_id column to files table for multi-user document isolation."""
+    op.add_column("files", sa.Column("owner_id", sa.String(255), nullable=True))
+    op.create_index("ix_files_owner_id", "files", ["owner_id"])
+
+
+def downgrade() -> None:
+    """Remove owner_id column from files table."""
+    op.drop_index("ix_files_owner_id", table_name="files")
+    op.drop_column("files", "owner_id")

tests/test_multi_user.py

@@ -0,0 +1,872 @@
+"""
+Tests for multi-user document isolation and feature flag.
+
+Covers:
+- user_scope utilities (get_current_owner_id, apply_owner_filter)
+- FileRecord.owner_id model field
+- API file list/detail/delete scoping in multi-user mode
+- Upload endpoint owner_id propagation
+- Feature flag toggling (single-user vs multi-user mode)
+"""
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+from sqlalchemy.pool import StaticPool
+
+from app.config import settings
+from app.database import Base
+from app.models import FileRecord
+
+# ---------------------------------------------------------------------------
+# Fixtures
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture()
+def mu_engine():
+    """In-memory SQLite engine for multi-user tests."""
+    engine = create_engine(
+        "sqlite:///:memory:",
+        connect_args={"check_same_thread": False},
+        poolclass=StaticPool,
+    )
+    Base.metadata.create_all(bind=engine)
+    yield engine
+    Base.metadata.drop_all(bind=engine)
+
+
+@pytest.fixture()
+def mu_session(mu_engine):
+    """Session scoped to a single test function."""
+    Session = sessionmaker(bind=mu_engine)
+    session = Session()
+    yield session
+    session.close()
+
+
+def _create_file_record(session, owner_id=None, filename="test.pdf"):
+    """Helper to insert a minimal FileRecord."""
+    rec = FileRecord(
+        filehash="abc123",
+        original_filename=filename,
+        local_filename="/tmp/test.pdf",
+        file_size=1024,
+        mime_type="application/pdf",
+        is_duplicate=False,
+        owner_id=owner_id,
+    )
+    session.add(rec)
+    session.commit()
+    session.refresh(rec)
+    return rec
+
+
+def _mock_request(user=None):
+    """Create a mock request with the given session user."""
+    request = MagicMock()
+    request.session = {"user": user} if user else {}
+    return request
+
+
+def _patch_multi_user(enabled):
+    """Return a patch context manager for multi_user_enabled."""
+    return patch.object(settings, "multi_user_enabled", enabled)
+
+
+# ---------------------------------------------------------------------------
+# Model tests
+# ---------------------------------------------------------------------------
+
+
+class TestFileRecordOwnerField:
+    """Verify the owner_id column on FileRecord."""
+
+    @pytest.mark.unit
+    def test_owner_id_defaults_to_none(self, mu_session):
+        """FileRecord created without owner_id should have None."""
+        rec = _create_file_record(mu_session)
+        assert rec.owner_id is None
+
+    @pytest.mark.unit
+    def test_owner_id_stores_value(self, mu_session):
+        """FileRecord created with owner_id should persist it."""
+        rec = _create_file_record(mu_session, owner_id="user@example.com")
+        assert rec.owner_id == "user@example.com"
+
+    @pytest.mark.unit
+    def test_owner_id_filterable(self, mu_session):
+        """Can query FileRecord by owner_id."""
+        _create_file_record(mu_session, owner_id="alice")
+        _create_file_record(mu_session, owner_id="bob")
+        _create_file_record(mu_session, owner_id=None)
+
+        alice_files = mu_session.query(FileRecord).filter(FileRecord.owner_id == "alice").all()
+        assert len(alice_files) == 1
+        assert alice_files[0].owner_id == "alice"
+
+        global_files = mu_session.query(FileRecord).filter(FileRecord.owner_id.is_(None)).all()
+        assert len(global_files) == 1
+
+
+# ---------------------------------------------------------------------------
+# user_scope utility tests
+# ---------------------------------------------------------------------------
+
+
+class TestGetCurrentOwnerId:
+    """Tests for get_current_owner_id()."""
+
+    @pytest.mark.unit
+    def test_returns_none_when_no_user(self):
+        from app.utils.user_scope import get_current_owner_id
+
+        request = _mock_request(user=None)
+        assert get_current_owner_id(request) is None
+
+    @pytest.mark.unit
+    def test_prefers_sub_claim(self):
+        from app.utils.user_scope import get_current_owner_id
+
+        request = _mock_request(user={"sub": "sub-123", "preferred_username": "alice", "email": "a@b.com"})
+        assert get_current_owner_id(request) == "sub-123"
+
+    @pytest.mark.unit
+    def test_falls_back_to_preferred_username(self):
+        from app.utils.user_scope import get_current_owner_id
+
+        request = _mock_request(user={"preferred_username": "alice", "email": "a@b.com"})
+        assert get_current_owner_id(request) == "alice"
+
+    @pytest.mark.unit
+    def test_falls_back_to_email(self):
+        from app.utils.user_scope import get_current_owner_id
+
+        request = _mock_request(user={"email": "a@b.com"})
+        assert get_current_owner_id(request) == "a@b.com"
+
+    @pytest.mark.unit
+    def test_falls_back_to_id(self):
+        from app.utils.user_scope import get_current_owner_id
+
+        request = _mock_request(user={"id": "admin"})
+        assert get_current_owner_id(request) == "admin"
+
+    @pytest.mark.unit
+    def test_returns_none_for_empty_session(self):
+        from app.utils.user_scope import get_current_owner_id
+
+        request = MagicMock()
+        request.session = {}
+        assert get_current_owner_id(request) is None
+
+
+class TestApplyOwnerFilter:
+    """Tests for apply_owner_filter()."""
+
+    @pytest.mark.unit
+    def test_no_filter_when_disabled(self, mu_session):
+        """When multi_user_enabled=False, all files are returned."""
+        from app.utils.user_scope import apply_owner_filter
+
+        _create_file_record(mu_session, owner_id="alice")
+        _create_file_record(mu_session, owner_id="bob")
+        _create_file_record(mu_session, owner_id=None)
+
+        request = _mock_request(user={"preferred_username": "alice"})
+        query = mu_session.query(FileRecord)
+
+        with _patch_multi_user(False):
+            filtered = apply_owner_filter(query, request)
+
+        assert filtered.count() == 3
+
+    @pytest.mark.unit
+    def test_filters_by_owner_when_enabled(self, mu_session):
+        """When multi_user_enabled=True with unowned_docs_visible, user sees own + unowned files."""
+        from app.utils.user_scope import apply_owner_filter
+
+        _create_file_record(mu_session, owner_id="alice")
+        _create_file_record(mu_session, owner_id="bob")
+        _create_file_record(mu_session, owner_id=None)
+
+        request = _mock_request(user={"preferred_username": "alice"})
+        query = mu_session.query(FileRecord)
+
+        with _patch_multi_user(True), patch.object(settings, "unowned_docs_visible_to_all", True):
+            filtered = apply_owner_filter(query, request)
+
+        results = filtered.all()
+        # Alice sees her own file + the unowned file (not Bob's)
+        assert len(results) == 2
+        owner_ids = {r.owner_id for r in results}
+        assert owner_ids == {"alice", None}
+
+    @pytest.mark.unit
+    def test_filters_strictly_when_unowned_not_visible(self, mu_session):
+        """When unowned_docs_visible_to_all=False, user sees only own files."""
+        from app.utils.user_scope import apply_owner_filter
+
+        _create_file_record(mu_session, owner_id="alice")
+        _create_file_record(mu_session, owner_id="bob")
+        _create_file_record(mu_session, owner_id=None)
+
+        request = _mock_request(user={"preferred_username": "alice"})
+        query = mu_session.query(FileRecord)
+
+        with _patch_multi_user(True), patch.object(settings, "unowned_docs_visible_to_all", False):
+            filtered = apply_owner_filter(query, request)
+
+        results = filtered.all()
+        assert len(results) == 1
+        assert results[0].owner_id == "alice"
+
+    @pytest.mark.unit
+    def test_admin_sees_all_when_enabled(self, mu_session):
+        """Admin users bypass the owner filter in multi-user mode."""
+        from app.utils.user_scope import apply_owner_filter
+
+        _create_file_record(mu_session, owner_id="alice")
+        _create_file_record(mu_session, owner_id="bob")
+        _create_file_record(mu_session, owner_id=None)
+
+        request = _mock_request(user={"preferred_username": "admin", "is_admin": True})
+        query = mu_session.query(FileRecord)
+
+        with _patch_multi_user(True):
+            filtered = apply_owner_filter(query, request)
+
+        assert filtered.count() == 3
+
+    @pytest.mark.unit
+    def test_unauthenticated_sees_nothing_when_enabled(self, mu_session):
+        """When no user is logged in and multi-user is enabled, return empty."""
+        from app.utils.user_scope import apply_owner_filter
+
+        _create_file_record(mu_session, owner_id="alice")
+
+        request = _mock_request(user=None)
+        query = mu_session.query(FileRecord)
+
+        with _patch_multi_user(True):
+            filtered = apply_owner_filter(query, request)
+
+        assert filtered.count() == 0
+
+
+# ---------------------------------------------------------------------------
+# Config / feature flag tests
+# ---------------------------------------------------------------------------
+
+
+class TestMultiUserConfig:
+    """Verify the multi-user configuration settings."""
+
+    @pytest.mark.unit
+    def test_multi_user_default_disabled(self):
+        """multi_user_enabled should default to False."""
+        from app.config import settings
+
+        # Default is False (overridable via env)
+        assert hasattr(settings, "multi_user_enabled")
+
+    @pytest.mark.unit
+    def test_default_daily_upload_limit_exists(self):
+        """default_daily_upload_limit should exist on settings."""
+        from app.config import settings
+
+        assert hasattr(settings, "default_daily_upload_limit")
+
+    @pytest.mark.unit
+    def test_multi_user_setting_has_metadata(self):
+        """multi_user_enabled must be in SETTING_METADATA."""
+        from app.utils.settings_service import SETTING_METADATA
+
+        assert "multi_user_enabled" in SETTING_METADATA
+        meta = SETTING_METADATA["multi_user_enabled"]
+        assert meta["type"] == "boolean"
+        assert meta["category"] == "Authentication"
+
+    @pytest.mark.unit
+    def test_daily_upload_limit_setting_has_metadata(self):
+        """default_daily_upload_limit must be in SETTING_METADATA."""
+        from app.utils.settings_service import SETTING_METADATA
+
+        assert "default_daily_upload_limit" in SETTING_METADATA
+        meta = SETTING_METADATA["default_daily_upload_limit"]
+        assert meta["type"] == "integer"
+
+
+# ---------------------------------------------------------------------------
+# Migration tests
+# ---------------------------------------------------------------------------
+
+
+class TestMigration012:
+    """Verify the multi-user migration file exists and is well-formed."""
+
+    @pytest.mark.unit
+    def test_migration_file_exists(self):
+        """Migration 012 should exist."""
+        from pathlib import Path
+
+        migration = Path("migrations/versions/012_add_multi_user_support.py")
+        assert migration.exists()
+
+    @pytest.mark.unit
+    def test_migration_chain(self):
+        """Migration 012 should chain from 011."""
+        import importlib.util
+
+        spec = importlib.util.spec_from_file_location(
+            "migration_012", "migrations/versions/012_add_multi_user_support.py"
+        )
+        mod = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(mod)
+        assert mod.down_revision == "011_add_pdfa_paths"
+        assert mod.revision == "012_add_multi_user_support"
+
+
+# ---------------------------------------------------------------------------
+# API integration tests (list files with owner scoping)
+# ---------------------------------------------------------------------------
+
+
+class TestFilesAPIMultiUser:
+    """Verify that the files API respects multi-user scoping."""
+
+    @pytest.mark.integration
+    def test_list_files_unscoped_single_user(self, client, db_session):
+        """In single-user mode all files are visible."""
+        _create_file_record(db_session, owner_id="alice", filename="a.pdf")
+        _create_file_record(db_session, owner_id="bob", filename="b.pdf")
+
+        with _patch_multi_user(False):
+            response = client.get("/api/files")
+
+        assert response.status_code == 200
+        data = response.json()
+        assert data["pagination"]["total"] == 2
+
+    @pytest.mark.integration
+    def test_list_files_scoped_multi_user(self, client, db_session):
+        """In multi-user mode only the user's files should be returned."""
+        _create_file_record(db_session, owner_id="alice", filename="a.pdf")
+        _create_file_record(db_session, owner_id="bob", filename="b.pdf")
+
+        with _patch_multi_user(True):
+            # Without a real session, the filter will return no results
+            # (unauthenticated user sees nothing in multi-user mode)
+            response = client.get("/api/files")
+            assert response.status_code == 200
+            data = response.json()
+            # No session user means empty results
+            assert data["pagination"]["total"] == 0
+
+    @pytest.mark.integration
+    def test_get_file_detail_respects_scope(self, client, db_session):
+        """File detail endpoint should return 404 for files owned by other users."""
+        rec = _create_file_record(db_session, owner_id="alice", filename="a.pdf")
+
+        with _patch_multi_user(True):
+            response = client.get(f"/api/files/{rec.id}")
+            # Without session, user is unauthenticated → 404
+            assert response.status_code == 404
+
+    @pytest.mark.integration
+    def test_get_file_detail_single_user_mode(self, client, db_session):
+        """File detail endpoint should work normally in single-user mode."""
+        rec = _create_file_record(db_session, owner_id="alice", filename="a.pdf")
+
+        with _patch_multi_user(False):
+            response = client.get(f"/api/files/{rec.id}")
+            assert response.status_code == 200
+            data = response.json()
+            assert data["file"]["id"] == rec.id
+
+
+# ---------------------------------------------------------------------------
+# process_document owner_id parameter tests
+# ---------------------------------------------------------------------------
+
+
+class TestProcessDocumentOwnerId:
+    """Verify process_document task accepts owner_id."""
+
+    @pytest.mark.unit
+    def test_process_document_signature_accepts_owner_id(self):
+        """process_document should accept owner_id as a keyword argument."""
+        import inspect
+
+        from app.tasks.process_document import process_document
+
+        sig = inspect.signature(process_document)
+        assert "owner_id" in sig.parameters
+        assert sig.parameters["owner_id"].default is None
+
+    @pytest.mark.unit
+    def test_convert_to_pdf_signature_accepts_owner_id(self):
+        """convert_to_pdf should accept owner_id as a keyword argument."""
+        import inspect
+
+        from app.tasks.convert_to_pdf import convert_to_pdf
+
+        sig = inspect.signature(convert_to_pdf)
+        assert "owner_id" in sig.parameters
+        assert sig.parameters["owner_id"].default is None
+
+    @pytest.mark.unit
+    def test_default_owner_id_fallback_in_process_document(self):
+        """When owner_id is None and default_owner_id is set, fallback is used."""
+        # Verify the fallback logic exists in the source code
+        import ast
+        from pathlib import Path
+
+        src = Path("app/tasks/process_document.py").read_text()
+        tree = ast.parse(src)
+        # Check that 'default_owner_id' string appears in the source
+        assert "default_owner_id" in src, "default_owner_id fallback not found in process_document.py"
+
+
+# ---------------------------------------------------------------------------
+# New config settings tests
+# ---------------------------------------------------------------------------
+
+
+class TestUnownedDocsConfig:
+    """Verify the new multi-user configuration settings."""
+
+    @pytest.mark.unit
+    def test_unowned_docs_visible_default_true(self):
+        """unowned_docs_visible_to_all should default to True."""
+        assert hasattr(settings, "unowned_docs_visible_to_all")
+
+    @pytest.mark.unit
+    def test_default_owner_id_default_none(self):
+        """default_owner_id should default to None."""
+        assert hasattr(settings, "default_owner_id")
+
+    @pytest.mark.unit
+    def test_unowned_docs_has_metadata(self):
+        """unowned_docs_visible_to_all must be in SETTING_METADATA."""
+        from app.utils.settings_service import SETTING_METADATA
+
+        assert "unowned_docs_visible_to_all" in SETTING_METADATA
+        meta = SETTING_METADATA["unowned_docs_visible_to_all"]
+        assert meta["type"] == "boolean"
+        assert meta["category"] == "Authentication"
+
+    @pytest.mark.unit
+    def test_default_owner_id_has_metadata(self):
+        """default_owner_id must be in SETTING_METADATA."""
+        from app.utils.settings_service import SETTING_METADATA
+
+        assert "default_owner_id" in SETTING_METADATA
+        meta = SETTING_METADATA["default_owner_id"]
+        assert meta["type"] == "user_autocomplete"
+
+
+# ---------------------------------------------------------------------------
+# Claim endpoint tests
+# ---------------------------------------------------------------------------
+
+
+class TestClaimEndpoint:
+    """Tests for POST /api/files/{file_id}/claim."""
+
+    @pytest.mark.integration
+    def test_claim_disabled_without_multi_user(self, client, db_session):
+        """Claiming is rejected when multi-user mode is off."""
+        rec = _create_file_record(db_session, owner_id=None, filename="unclaimed.pdf")
+        with _patch_multi_user(False):
+            response = client.post(f"/api/files/{rec.id}/claim")
+        assert response.status_code == 400
+        assert "not enabled" in response.json()["detail"]
+
+    @pytest.mark.integration
+    def test_claim_unauthenticated_returns_401(self, client, db_session):
+        """Claiming without a session returns 401."""
+        rec = _create_file_record(db_session, owner_id=None, filename="unclaimed.pdf")
+        with _patch_multi_user(True):
+            response = client.post(f"/api/files/{rec.id}/claim")
+        assert response.status_code == 401
+
+
+class TestClaimUnit:
+    """Unit tests for claim logic directly on the model."""
+
+    @pytest.mark.unit
+    def test_claim_sets_owner_id(self, mu_session):
+        """Setting owner_id on a NULL-owner file persists correctly."""
+        rec = _create_file_record(mu_session, owner_id=None)
+        assert rec.owner_id is None
+
+        rec.owner_id = "alice"
+        mu_session.commit()
+        mu_session.refresh(rec)
+        assert rec.owner_id == "alice"
+
+    @pytest.mark.unit
+    def test_claim_already_owned_same_user(self, mu_session):
+        """Claiming a file you already own returns already_owned status."""
+        rec = _create_file_record(mu_session, owner_id="alice")
+        # The API would return already_owned; test the model constraint
+        assert rec.owner_id == "alice"
+
+    @pytest.mark.unit
+    def test_claim_unowned_then_verify(self, mu_session):
+        """Claim unowned, then verify the file is filterable by new owner."""
+        rec = _create_file_record(mu_session, owner_id=None, filename="orphan.pdf")
+        # Before claim: found via NULL filter
+        unowned = mu_session.query(FileRecord).filter(FileRecord.owner_id.is_(None)).all()
+        assert rec in unowned
+
+        # After claim
+        rec.owner_id = "claimer"
+        mu_session.commit()
+        mu_session.refresh(rec)
+
+        owned = mu_session.query(FileRecord).filter(FileRecord.owner_id == "claimer").all()
+        assert rec in owned
+        unowned = mu_session.query(FileRecord).filter(FileRecord.owner_id.is_(None)).all()
+        assert rec not in unowned
+
+
+# ---------------------------------------------------------------------------
+# Bulk claim endpoint tests
+# ---------------------------------------------------------------------------
+
+
+class TestBulkClaimEndpoint:
+    """Tests for POST /api/files/bulk-claim."""
+
+    @pytest.mark.integration
+    def test_bulk_claim_disabled_without_multi_user(self, client, db_session):
+        """Bulk claiming is rejected when multi-user mode is off."""
+        _create_file_record(db_session, owner_id=None, filename="a.pdf")
+        with _patch_multi_user(False):
+            response = client.post("/api/files/bulk-claim", json=[1])
+        assert response.status_code == 400
+
+    @pytest.mark.integration
+    def test_bulk_claim_unauthenticated_returns_401(self, client, db_session):
+        """Bulk claiming without a session returns 401."""
+        rec = _create_file_record(db_session, owner_id=None, filename="a.pdf")
+        with _patch_multi_user(True):
+            response = client.post("/api/files/bulk-claim", json=[rec.id])
+        assert response.status_code == 401
+
+
+class TestBulkClaimUnit:
+    """Unit tests for bulk claim DB logic."""
+
+    @pytest.mark.unit
+    def test_bulk_claim_multiple_unowned(self, mu_session):
+        """Bulk claim sets owner on multiple NULL-owner files."""
+        r1 = _create_file_record(mu_session, owner_id=None, filename="a.pdf")
+        r2 = _create_file_record(mu_session, owner_id=None, filename="b.pdf")
+        r3 = _create_file_record(mu_session, owner_id="bob", filename="c.pdf")
+
+        # Simulate bulk claim logic
+        records = mu_session.query(FileRecord).filter(FileRecord.id.in_([r1.id, r2.id, r3.id])).all()
+        claimed, skipped = [], []
+        for rec in records:
+            if rec.owner_id is None:
+                rec.owner_id = "claimer"
+                claimed.append(rec.id)
+            else:
+                skipped.append(rec.id)
+        mu_session.commit()
+
+        assert set(claimed) == {r1.id, r2.id}
+        assert skipped == [r3.id]
+
+    @pytest.mark.unit
+    def test_bulk_claim_all_already_owned(self, mu_session):
+        """Bulk claim with all already-owned files skips everything."""
+        r1 = _create_file_record(mu_session, owner_id="alice", filename="a.pdf")
+        r2 = _create_file_record(mu_session, owner_id="bob", filename="b.pdf")
+
+        records = mu_session.query(FileRecord).filter(FileRecord.id.in_([r1.id, r2.id])).all()
+        claimed = [rec for rec in records if rec.owner_id is None]
+        assert len(claimed) == 0
+
+
+# ---------------------------------------------------------------------------
+# Assign-owner endpoint tests
+# ---------------------------------------------------------------------------
+
+
+class TestAssignOwnerEndpoint:
+    """Tests for POST /api/files/assign-owner."""
+
+    @pytest.mark.integration
+    def test_assign_owner_disabled_without_multi_user(self, client, db_session):
+        """Assigning owner is rejected when multi-user mode is off."""
+        with _patch_multi_user(False):
+            response = client.post("/api/files/assign-owner?owner_id=alice")
+        assert response.status_code == 400
+
+    @pytest.mark.integration
+    def test_assign_owner_requires_admin(self, client, db_session):
+        """Non-admin users cannot assign owners."""
+        with _patch_multi_user(True):
+            response = client.post("/api/files/assign-owner?owner_id=alice")
+        # 403 (non-admin) or 401 (no auth)
+        assert response.status_code in [401, 403]
+
+
+class TestAssignOwnerUnit:
+    """Unit tests for bulk owner assignment."""
+
+    @pytest.mark.unit
+    def test_assign_owner_to_unowned_files(self, mu_session):
+        """Bulk update sets owner_id on all NULL-owner files."""
+        _create_file_record(mu_session, owner_id=None, filename="a.pdf")
+        _create_file_record(mu_session, owner_id=None, filename="b.pdf")
+        _create_file_record(mu_session, owner_id="bob", filename="c.pdf")
+
+        updated = (
+            mu_session.query(FileRecord)
+            .filter(FileRecord.owner_id.is_(None))
+            .update({FileRecord.owner_id: "alice"}, synchronize_session="fetch")
+        )
+        mu_session.commit()
+
+        assert updated == 2
+        all_files = mu_session.query(FileRecord).all()
+        owners = {f.original_filename: f.owner_id for f in all_files}
+        assert owners["a.pdf"] == "alice"
+        assert owners["b.pdf"] == "alice"
+        assert owners["c.pdf"] == "bob"
+
+    @pytest.mark.unit
+    def test_assign_owner_to_specific_files(self, mu_session):
+        """Update specific file IDs sets owner_id."""
+        rec1 = _create_file_record(mu_session, owner_id=None, filename="a.pdf")
+        rec2 = _create_file_record(mu_session, owner_id="bob", filename="b.pdf")
+
+        updated = (
+            mu_session.query(FileRecord)
+            .filter(FileRecord.id.in_([rec1.id, rec2.id]))
+            .update({FileRecord.owner_id: "charlie"}, synchronize_session="fetch")
+        )
+        mu_session.commit()
+
+        assert updated == 2
+        mu_session.refresh(rec1)
+        mu_session.refresh(rec2)
+        assert rec1.owner_id == "charlie"
+        assert rec2.owner_id == "charlie"
+
+
+# ---------------------------------------------------------------------------
+# User search endpoint tests
+# ---------------------------------------------------------------------------
+
+
+class TestUserSearchEndpoint:
+    """Tests for GET /api/users/search."""
+
+    @pytest.mark.integration
+    def test_search_returns_known_users(self, client, db_session):
+        """Search should return distinct owner_ids from file records."""
+        _create_file_record(db_session, owner_id="alice", filename="a.pdf")
+        _create_file_record(db_session, owner_id="bob", filename="b.pdf")
+        _create_file_record(db_session, owner_id="alice", filename="a2.pdf")  # duplicate owner
+        _create_file_record(db_session, owner_id=None, filename="c.pdf")  # unowned
+
+        response = client.get("/api/users/search?q=")
+        assert response.status_code == 200
+        data = response.json()
+        assert "users" in data
+        # Should contain alice and bob (not None)
+        assert set(data["users"]) == {"alice", "bob"}
+
+    @pytest.mark.integration
+    def test_search_filters_by_substring(self, client, db_session):
+        """Search should filter by case-insensitive substring."""
+        _create_file_record(db_session, owner_id="christianlouis", filename="a.pdf")
+        _create_file_record(db_session, owner_id="bob", filename="b.pdf")
+        _create_file_record(db_session, owner_id="alice", filename="c.pdf")
+
+        response = client.get("/api/users/search?q=risti")
+        assert response.status_code == 200
+        data = response.json()
+        assert data["users"] == ["christianlouis"]
+
+    @pytest.mark.integration
+    def test_search_respects_limit(self, client, db_session):
+        """Search should respect the limit parameter."""
+        for i in range(10):
+            _create_file_record(db_session, owner_id=f"user_{i:02d}", filename=f"file_{i}.pdf")
+
+        response = client.get("/api/users/search?q=user&limit=3")
+        assert response.status_code == 200
+        data = response.json()
+        assert len(data["users"]) == 3
+
+    @pytest.mark.integration
+    def test_search_empty_when_no_matches(self, client, db_session):
+        """Search with no matches should return empty list."""
+        _create_file_record(db_session, owner_id="alice", filename="a.pdf")
+
+        response = client.get("/api/users/search?q=zzzzz")
+        assert response.status_code == 200
+        data = response.json()
+        assert data["users"] == []
+
+    @pytest.mark.integration
+    def test_search_case_insensitive(self, client, db_session):
+        """Search should be case-insensitive."""
+        _create_file_record(db_session, owner_id="ChristianLouis", filename="a.pdf")
+
+        response = client.get("/api/users/search?q=CHRISTIAN")
+        assert response.status_code == 200
+        data = response.json()
+        assert data["users"] == ["ChristianLouis"]
+
+    @pytest.mark.integration
+    def test_search_no_query_param(self, client, db_session):
+        """Search without q parameter defaults to empty string (returns all)."""
+        _create_file_record(db_session, owner_id="alice", filename="a.pdf")
+        _create_file_record(db_session, owner_id="bob", filename="b.pdf")
+
+        response = client.get("/api/users/search")
+        assert response.status_code == 200
+        data = response.json()
+        assert set(data["users"]) == {"alice", "bob"}
+
+    @pytest.mark.integration
+    def test_search_empty_database(self, client, db_session):
+        """Search on an empty database returns empty list."""
+        response = client.get("/api/users/search?q=anything")
+        assert response.status_code == 200
+        data = response.json()
+        assert data["users"] == []
+
+    @pytest.mark.integration
+    def test_search_results_sorted_alphabetically(self, client, db_session):
+        """Search results should be sorted alphabetically."""
+        _create_file_record(db_session, owner_id="zebra", filename="z.pdf")
+        _create_file_record(db_session, owner_id="apple", filename="a.pdf")
+        _create_file_record(db_session, owner_id="mango", filename="m.pdf")
+
+        response = client.get("/api/users/search?q=")
+        assert response.status_code == 200
+        data = response.json()
+        assert data["users"] == ["apple", "mango", "zebra"]
+
+    @pytest.mark.integration
+    def test_search_default_limit_is_5(self, client, db_session):
+        """Default limit should be 5."""
+        for i in range(10):
+            _create_file_record(db_session, owner_id=f"user_{i:02d}", filename=f"file_{i}.pdf")
+
+        response = client.get("/api/users/search?q=user")
+        assert response.status_code == 200
+        data = response.json()
+        assert len(data["users"]) == 5
+
+    @pytest.mark.integration
+    def test_search_whitespace_query_trimmed(self, client, db_session):
+        """Leading/trailing whitespace in query should be trimmed."""
+        _create_file_record(db_session, owner_id="alice", filename="a.pdf")
+
+        response = client.get("/api/users/search?q=%20alice%20")
+        assert response.status_code == 200
+        data = response.json()
+        assert data["users"] == ["alice"]
+
+
+# ---------------------------------------------------------------------------
+# Additional user_scope filter edge-case tests
+# ---------------------------------------------------------------------------
+
+
+class TestApplyOwnerFilterEdgeCases:
+    """Additional edge-case tests for apply_owner_filter."""
+
+    @pytest.mark.unit
+    def test_unowned_visible_includes_null_and_own(self, mu_session):
+        """With unowned_docs_visible_to_all=True, user sees own + NULL files."""
+        from app.utils.user_scope import apply_owner_filter
+
+        own = _create_file_record(mu_session, owner_id="alice", filename="mine.pdf")
+        other = _create_file_record(mu_session, owner_id="bob", filename="theirs.pdf")
+        orphan = _create_file_record(mu_session, owner_id=None, filename="orphan.pdf")
+
+        request = _mock_request(user={"preferred_username": "alice"})
+        query = mu_session.query(FileRecord)
+
+        with _patch_multi_user(True), patch.object(settings, "unowned_docs_visible_to_all", True):
+            results = apply_owner_filter(query, request).all()
+
+        result_ids = {r.id for r in results}
+        assert own.id in result_ids
+        assert orphan.id in result_ids
+        assert other.id not in result_ids
+
+    @pytest.mark.unit
+    def test_unowned_not_visible_excludes_null(self, mu_session):
+        """With unowned_docs_visible_to_all=False, user sees only own files."""
+        from app.utils.user_scope import apply_owner_filter
+
+        own = _create_file_record(mu_session, owner_id="alice", filename="mine.pdf")
+        orphan = _create_file_record(mu_session, owner_id=None, filename="orphan.pdf")
+
+        request = _mock_request(user={"preferred_username": "alice"})
+        query = mu_session.query(FileRecord)
+
+        with _patch_multi_user(True), patch.object(settings, "unowned_docs_visible_to_all", False):
+            results = apply_owner_filter(query, request).all()
+
+        result_ids = {r.id for r in results}
+        assert own.id in result_ids
+        assert orphan.id not in result_ids
+
+    @pytest.mark.unit
+    def test_admin_always_sees_all(self, mu_session):
+        """Admin user always sees all files, regardless of unowned_docs_visible_to_all."""
+        from app.utils.user_scope import apply_owner_filter
+
+        _create_file_record(mu_session, owner_id="alice")
+        _create_file_record(mu_session, owner_id=None)
+
+        request = _mock_request(user={"preferred_username": "admin", "is_admin": True})
+        query = mu_session.query(FileRecord)
+
+        with _patch_multi_user(True), patch.object(settings, "unowned_docs_visible_to_all", False):
+            results = apply_owner_filter(query, request).all()
+
+        assert len(results) == 2
+
+    @pytest.mark.unit
+    def test_no_files_returns_empty(self, mu_session):
+        """Empty database returns no results for any user."""
+        from app.utils.user_scope import apply_owner_filter
+
+        request = _mock_request(user={"preferred_username": "alice"})
+        query = mu_session.query(FileRecord)
+
+        with _patch_multi_user(True):
+            results = apply_owner_filter(query, request).all()
+
+        assert results == []
+
+    @pytest.mark.unit
+    def test_user_with_sub_claim(self, mu_session):
+        """User with OAuth 'sub' claim uses sub as owner_id."""
+        from app.utils.user_scope import apply_owner_filter
+
+        rec = _create_file_record(mu_session, owner_id="oauth-sub-123")
+        request = _mock_request(user={"sub": "oauth-sub-123", "preferred_username": "alice"})
+        query = mu_session.query(FileRecord)
+
+        with _patch_multi_user(True), patch.object(settings, "unowned_docs_visible_to_all", False):
+            results = apply_owner_filter(query, request).all()
+
+        assert len(results) == 1
+        assert results[0].id == rec.id

tests/test_original_filename_preservation.py

@@ -95,6 +95,10 @@ startxref
     ):
         # Setup mocks
         mock_settings.workdir = str(tmp_path)
+        mock_settings.default_owner_id = ""
+        mock_settings.enable_deduplication = False
+        mock_settings.show_deduplication_step = False
+        mock_settings.enable_text_quality_check = False
         mock_session_local.return_value.__enter__.return_value = db_session
         mock_session_local.return_value.__exit__.return_value = None
         mock_extract.delay = MagicMock()
@@ -195,6 +199,10 @@ startxref
     ):
         # Setup mocks
         mock_settings.workdir = str(tmp_path)
+        mock_settings.default_owner_id = ""
+        mock_settings.enable_deduplication = False
+        mock_settings.show_deduplication_step = False
+        mock_settings.enable_text_quality_check = False
         mock_session_local.return_value.__enter__.return_value = db_session
         mock_session_local.return_value.__exit__.return_value = None
         mock_extract.delay = MagicMock()

tests/test_suggestion_providers.py

@@ -0,0 +1,299 @@
+"""
+Tests for the suggestion providers and the settings suggestions API.
+
+Covers:
+- Dynamic suggestion providers (AWS, Azure, Tesseract, EasyOCR, embedding models)
+- GET /api/settings/{key}/suggestions endpoint
+- Substring filtering and limit enforcement
+- Fallback to static lists when SDKs are unavailable
+"""
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from app.utils.suggestion_providers import (
+    _AZURE_REGIONS_STATIC,
+    _EASYOCR_LANGS_STATIC,
+    _EMBEDDING_MODELS,
+    _TESSERACT_LANGS_STATIC,
+    SUGGESTION_PROVIDERS,
+    get_aws_regions,
+    get_azure_regions,
+    get_easyocr_languages,
+    get_embedding_models,
+    get_suggestions,
+    get_tesseract_languages,
+)
+
+# ---------------------------------------------------------------------------
+# Provider unit tests
+# ---------------------------------------------------------------------------
+
+
+class TestAWSRegionProvider:
+    """Tests for get_aws_regions."""
+
+    @pytest.mark.unit
+    def test_returns_list(self):
+        """Should return a non-empty list of region strings."""
+        regions = get_aws_regions()
+        assert isinstance(regions, list)
+        assert len(regions) > 0
+        assert all(isinstance(r, str) for r in regions)
+
+    @pytest.mark.unit
+    def test_us_east_1_present(self):
+        """us-east-1 should always be in the list."""
+        regions = get_aws_regions()
+        assert "us-east-1" in regions
+
+    @pytest.mark.unit
+    def test_results_are_sorted(self):
+        """Region list should be sorted alphabetically."""
+        regions = get_aws_regions()
+        assert regions == sorted(regions)
+
+    @pytest.mark.unit
+    def test_fallback_on_boto3_failure(self):
+        """Should fall back to static list when boto3 raises."""
+        with patch.dict("sys.modules", {"boto3": None}):
+            regions = get_aws_regions()
+        # Should still return a list (the static fallback)
+        assert isinstance(regions, list)
+        assert "us-east-1" in regions
+
+
+class TestAzureRegionProvider:
+    """Tests for get_azure_regions."""
+
+    @pytest.mark.unit
+    def test_returns_static_list(self):
+        """Should return the curated static list."""
+        regions = get_azure_regions()
+        assert regions == _AZURE_REGIONS_STATIC
+        assert "eastus" in regions
+
+    @pytest.mark.unit
+    def test_contains_common_regions(self):
+        """Common Azure regions should be present."""
+        regions = get_azure_regions()
+        for region in ["eastus", "westeurope", "uksouth", "japaneast"]:
+            assert region in regions
+
+
+class TestTesseractLanguageProvider:
+    """Tests for get_tesseract_languages."""
+
+    @pytest.mark.unit
+    def test_returns_list(self):
+        """Should return a non-empty list."""
+        langs = get_tesseract_languages()
+        assert isinstance(langs, list)
+        assert len(langs) > 0
+
+    @pytest.mark.unit
+    def test_eng_present(self):
+        """English should always be available."""
+        langs = get_tesseract_languages()
+        assert "eng" in langs
+
+    @pytest.mark.unit
+    def test_fallback_on_missing_tesseract(self):
+        """Should fall back to static list when tesseract is not installed."""
+        with patch("subprocess.run", side_effect=FileNotFoundError):
+            langs = get_tesseract_languages()
+        assert langs == _TESSERACT_LANGS_STATIC
+
+    @pytest.mark.unit
+    def test_uses_subprocess_output_when_available(self):
+        """Should parse subprocess output when tesseract is installed."""
+        mock_result = MagicMock()
+        mock_result.returncode = 0
+        mock_result.stdout = "List of available languages (4):\neng\ndeu\nfra\nita\n"
+
+        with patch("subprocess.run", return_value=mock_result):
+            langs = get_tesseract_languages()
+
+        assert langs == ["deu", "eng", "fra", "ita"]
+
+
+class TestEasyOCRLanguageProvider:
+    """Tests for get_easyocr_languages."""
+
+    @pytest.mark.unit
+    def test_returns_list(self):
+        """Should return a non-empty list."""
+        langs = get_easyocr_languages()
+        assert isinstance(langs, list)
+        assert len(langs) > 0
+
+    @pytest.mark.unit
+    def test_en_present(self):
+        """English should always be available."""
+        langs = get_easyocr_languages()
+        assert "en" in langs
+
+    @pytest.mark.unit
+    def test_fallback_when_easyocr_missing(self):
+        """Should fall back to static list when easyocr is not installed."""
+        # easyocr is not installed in the test env, so this tests the real fallback
+        langs = get_easyocr_languages()
+        assert langs == _EASYOCR_LANGS_STATIC
+
+
+class TestEmbeddingModelProvider:
+    """Tests for get_embedding_models."""
+
+    @pytest.mark.unit
+    def test_returns_list(self):
+        """Should return a non-empty list."""
+        models = get_embedding_models()
+        assert isinstance(models, list)
+        assert len(models) > 0
+
+    @pytest.mark.unit
+    def test_default_model_present(self):
+        """The default model should be in the list."""
+        models = get_embedding_models()
+        assert "text-embedding-3-small" in models
+
+    @pytest.mark.unit
+    def test_returns_static_list(self):
+        """Should return the static embedding model list."""
+        assert get_embedding_models() == _EMBEDDING_MODELS
+
+
+# ---------------------------------------------------------------------------
+# get_suggestions() tests
+# ---------------------------------------------------------------------------
+
+
+class TestGetSuggestions:
+    """Tests for the get_suggestions aggregator function."""
+
+    @pytest.mark.unit
+    def test_all_providers_registered(self):
+        """All expected keys should be registered."""
+        expected_keys = {"aws_region", "azure_region", "tesseract_language", "easyocr_languages", "embedding_model"}
+        assert expected_keys == set(SUGGESTION_PROVIDERS.keys())
+
+    @pytest.mark.unit
+    def test_unregistered_key_raises(self):
+        """Requesting suggestions for an unknown key raises KeyError."""
+        with pytest.raises(KeyError, match="no_such_setting"):
+            get_suggestions("no_such_setting")
+
+    @pytest.mark.unit
+    def test_empty_query_returns_all(self):
+        """Empty query returns all suggestions up to the limit."""
+        results = get_suggestions("embedding_model", query="", limit=100)
+        assert len(results) == len(_EMBEDDING_MODELS)
+
+    @pytest.mark.unit
+    def test_substring_filtering(self):
+        """Query filters by case-insensitive substring."""
+        results = get_suggestions("aws_region", query="east", limit=50)
+        assert all("east" in r.lower() for r in results)
+        assert len(results) > 0
+
+    @pytest.mark.unit
+    def test_case_insensitive(self):
+        """Filtering should be case-insensitive."""
+        results = get_suggestions("aws_region", query="EAST", limit=50)
+        assert len(results) > 0
+        assert "us-east-1" in results
+
+    @pytest.mark.unit
+    def test_limit_respected(self):
+        """Results should not exceed the limit."""
+        results = get_suggestions("tesseract_language", query="", limit=3)
+        assert len(results) == 3
+
+    @pytest.mark.unit
+    def test_whitespace_trimmed(self):
+        """Leading/trailing whitespace in the query should be trimmed."""
+        results = get_suggestions("aws_region", query="  us-east  ", limit=10)
+        assert all("us-east" in r.lower() for r in results)
+
+
+# ---------------------------------------------------------------------------
+# API endpoint integration tests
+# ---------------------------------------------------------------------------
+
+
+class TestSuggestionsEndpoint:
+    """Tests for GET /api/settings/{key}/suggestions."""
+
+    @pytest.mark.integration
+    def test_aws_region_suggestions(self, client):
+        """AWS region endpoint returns suggestions."""
+        response = client.get("/api/settings/aws_region/suggestions?q=east")
+        assert response.status_code == 200
+        data = response.json()
+        assert "suggestions" in data
+        assert data["key"] == "aws_region"
+        assert len(data["suggestions"]) > 0
+        assert all("east" in s.lower() for s in data["suggestions"])
+
+    @pytest.mark.integration
+    def test_azure_region_suggestions(self, client):
+        """Azure region endpoint returns suggestions."""
+        response = client.get("/api/settings/azure_region/suggestions?q=europe")
+        assert response.status_code == 200
+        data = response.json()
+        assert "westeurope" in data["suggestions"]
+
+    @pytest.mark.integration
+    def test_tesseract_language_suggestions(self, client):
+        """Tesseract language endpoint returns suggestions."""
+        response = client.get("/api/settings/tesseract_language/suggestions?q=eng")
+        assert response.status_code == 200
+        data = response.json()
+        assert "eng" in data["suggestions"]
+
+    @pytest.mark.integration
+    def test_easyocr_language_suggestions(self, client):
+        """EasyOCR language endpoint returns suggestions."""
+        response = client.get("/api/settings/easyocr_languages/suggestions?q=de")
+        assert response.status_code == 200
+        data = response.json()
+        assert "de" in data["suggestions"]
+
+    @pytest.mark.integration
+    def test_embedding_model_suggestions(self, client):
+        """Embedding model endpoint returns suggestions."""
+        response = client.get("/api/settings/embedding_model/suggestions?q=embed")
+        assert response.status_code == 200
+        data = response.json()
+        assert len(data["suggestions"]) > 0
+
+    @pytest.mark.integration
+    def test_unknown_key_returns_404(self, client):
+        """Unknown setting key returns 404."""
+        response = client.get("/api/settings/nonexistent_setting/suggestions")
+        assert response.status_code == 404
+
+    @pytest.mark.integration
+    def test_empty_query_returns_results(self, client):
+        """Empty query returns all suggestions up to limit."""
+        response = client.get("/api/settings/embedding_model/suggestions?q=")
+        assert response.status_code == 200
+        data = response.json()
+        assert len(data["suggestions"]) > 0
+
+    @pytest.mark.integration
+    def test_limit_parameter(self, client):
+        """Limit parameter restricts the number of results."""
+        response = client.get("/api/settings/aws_region/suggestions?q=&limit=3")
+        assert response.status_code == 200
+        data = response.json()
+        assert len(data["suggestions"]) <= 3
+
+    @pytest.mark.integration
+    def test_limit_clamped_to_max(self, client):
+        """Limit over 50 should be clamped to 50."""
+        response = client.get("/api/settings/tesseract_language/suggestions?q=&limit=999")
+        assert response.status_code == 200
+        data = response.json()
+        assert len(data["suggestions"]) <= 50