"""
Document Service - Main orchestrator for document management.
Uses SQLAlchemy ORM through DocumentDbService instead of raw SQL.
Refactored from document_base.py DocumentAssistant class.

Supports both:
- SQLAlchemy Session mode (for FastAPI dependency injection)
- db_file mode (for agent-specific databases, backward compatible with DocumentAssistant)
"""
import os
import io
import json
import hashlib
import re
import shutil
import uuid
from datetime import datetime
from pathlib import Path
from typing import Any, List, Dict, Optional, Tuple, Union
from concurrent.futures import ThreadPoolExecutor, as_completed

from sqlalchemy.orm import Session
from openai import OpenAI

from app.services.document_db_service import DocumentDbService
from app.utils.content_extractor import ContentExtractor
from app.utils.image_extractor import ImageExtractor
from app.utils.chroma_vector_db import ChromaVectorDB
from app.utils.ai_analyzer import AIAnalyzer
from app.services.document_nodes import (
    timer,
    BM25Index,
    DocumentChunk,
    DocumentRAGPipeline,
    DocumentRAGConfig,
)
from app.services.langfuse_client import langfuse as lf_client
from app.utils.document_preview import (
    DEFAULT_PREVIEW_PDF_CACHE_DIR,
    apply_page_download_url_to_mapping,
    apply_page_image_url_to_mapping,
    apply_preview_contract_to_mapping,
    build_matching_page_urls,
    build_page_images_from_api_paths,
)
from app.utils.document_ingestion import (
    OFFICE_EXTENSIONS,
    extract_page_texts_for_path,
    prepare_ingestion,
    preview_pdf_path,
)
from app.utils.page_snapshot_mapping import (
    join_page_texts,
    page_number_from_metadata,
)
from app.utils.retrieval_scoring import chunk_relevance_scores_0_100
from app.utils.file_upload_check import (
    check_file_enabled_from_env_or_settings,
    evaluate_check_file_upload,
)
from app.core.config import settings

# Max ranked chunks returned as chunk_hits (primary search contract).
TOP_CHUNK_HITS = 10
from app.utils.file_schedule_utils import clear_upload_scheduling_for_paths
from app.services.folder_scan_job_prompts import (
    FOLDER_SCAN_JOB_COMPLETED,
    FOLDER_SCAN_JOB_FAILED,
    FOLDER_SCAN_JOB_INVALID_PATH,
    FOLDER_SCAN_JOB_NOT_FOUND,
    FOLDER_SCAN_JOB_STARTED,
    FOLDER_SCAN_JOB_STEP_PROGRESS,
    FOLDER_SCAN_JOB_STOPPED,
)

# Shared store for all agents with FILE datasource (single SQLite + Chroma + disk roots).
FILE_SYS_METADATA_DB = "file_sys_metadata.db"
FILE_VECTORS_DB_DIR = "file_vectors_db"
FILE_UPLOAD_DISK_ROOT = "uploaded_files_/file_data"
FILE_IMAGE_DISK_ROOT = "document_images_/file_data"
FILE_EXTRACT_DISK_ROOT = "extracted_images_/file_data"


def enrich_document_search_file_paths_from_db(
    doc_results: Dict[str, Any],
    *,
    db_file: str = FILE_SYS_METADATA_DB,
    upload_folder: str = FILE_UPLOAD_DISK_ROOT,
    agent_id: Optional[Union[str, List[str]]] = None,
) -> None:
    """Resolve file_path on document_search (@file) payloads from files.file_path in metadata DB."""
    if not doc_results or doc_results.get("type") != "document_search":
        return
    try:
        svc = DocumentService(
            db_file=db_file,
            upload_folder=upload_folder,
            image_folder=FILE_IMAGE_DISK_ROOT,
            extracted_folder=FILE_EXTRACT_DISK_ROOT,
            agent_id=agent_id,
        )
    except Exception:
        return
    svc._enrich_hits_file_paths(doc_results.get("chunk_hits"))
    svc._enrich_hits_file_paths(doc_results.get("results"))
    svc._enrich_hits_preview_contract(doc_results.get("chunk_hits"))
    svc._enrich_hits_preview_contract(doc_results.get("results"))
    folder_groups = doc_results.get("folder_groups") or {}
    if isinstance(folder_groups, dict):
        for items in folder_groups.values():
            if isinstance(items, list):
                svc._enrich_hits_file_paths(items)
                svc._enrich_hits_preview_contract(items)


class DocumentService:
    """
    Main service for document management operations.
    Uses SQLAlchemy ORM for database operations via DocumentDbService.
    
    This replaces DocumentAssistant from document_base.py with FastAPI-compatible
    session-based database access, while also supporting agent-specific databases.
    """
    
    def __init__(self, 
                 db: Union[Session, str] = None,
                 upload_folder: str = "uploaded_files",
                 image_folder: str = "document_images",
                 extracted_folder: str = "extracted_images",
                 db_file: str = None,
                 chroma_path: str = "file_vectors_db",
                 openai_api_key: str = None,
                 claude_api_key: str = None,
                  **kwargs):
        """
        Initialize DocumentService with SQLAlchemy session or database file.
        
        Args:
            db: SQLAlchemy Session from FastAPI dependency injection OR db_file path
            upload_folder: Folder for uploaded files
            image_folder: Folder for document preview images
            extracted_folder: Folder for extracted images from documents
            db_file: Database file path (for backward compatibility with DocumentAssistant)
            chroma_path: Path for ChromaDB vector storage
            openai_api_key: OpenAI API key for embeddings and optional chat fallback
            claude_api_key: Anthropic key when use_claude is enabled (chat via LLMService)
            agent_id: Agent ID string
        """
        # Create folders
        Path(extracted_folder).mkdir(parents=True, exist_ok=True)
        
        self.upload_folder = upload_folder
        self.image_folder = image_folder
        self.extracted_folder = extracted_folder
        self.agent_id = kwargs.get('agent_id', None)
        if isinstance(self.agent_id, list):
            self.agent_id = [str(aid).strip() for aid in self.agent_id if str(aid).strip()]
        elif self.agent_id:
            self.agent_id = str(self.agent_id).strip()
        
        # Support multiple initialization modes
        if db_file is not None:
            # db_file parameter provided (backward compatible mode)
            self.db_service = DocumentDbService(db_file=db_file, agent_id=self.agent_id)
        elif isinstance(db, str):
            # db is a string path
            self.db_service = DocumentDbService(db_file=db, agent_id=self.agent_id)
        elif db is not None:
            # db is a SQLAlchemy Session
            self.db_service = DocumentDbService(db, agent_id=self.agent_id)
        else:
            raise ValueError("Either db (Session) or db_file must be provided")
        
        self.content_extractor = ContentExtractor()
        self.image_extractor = ImageExtractor(extracted_folder)
        self.vector_db = ChromaVectorDB(chroma_path, openai_api_key) if openai_api_key else None
        self.bm25_index = (
            BM25Index.for_chroma_path(chroma_path, openai_api_key=openai_api_key)
            if openai_api_key
            else None
        )
        self.ai_analyzer = (
            AIAnalyzer(openai_api_key or "", claude_api_key or "")
            if openai_api_key
            else None
        )
        self.openai_client = OpenAI(api_key=openai_api_key) if openai_api_key else None

    def _content_from_saved_path(
        self,
        file_path: str,
        file_ext: str,
        *,
        should_cancel=None,
    ) -> str:
        """Full text for hashing and DB preview from canonical PDF page texts."""
        ext = (file_ext or "").lower()
        if ext in OFFICE_EXTENSIONS | {"pdf"}:
            page_texts = extract_page_texts_for_path(
                file_path, ext, should_cancel=should_cancel
            )
            full_text, _ = join_page_texts(page_texts)
            text = (full_text or "").strip()
            return text if text else "No text extracted from document"
        return ""

    def _index_chunks_dense_and_sparse(self, ids: List[str], documents: List[str], metadatas: List[dict]) -> None:
        print(f"[FILE] Indexing chunks: dense+bm25, count={len(ids)}")
        if self.bm25_index:
            self.bm25_index.add_chunks(ids, documents, metadatas=metadatas)

    def _persistent_upload_path(self, file_id: str, file_extension: str) -> str:
        """Canonical on-disk path for an uploaded file copy under upload_folder."""
        ext = (file_extension or "bin").lower().lstrip(".")
        Path(self.upload_folder).mkdir(parents=True, exist_ok=True)
        return os.path.join(self.upload_folder, f"{file_id}.{ext}")

    def _copy_to_persistent_upload(
        self, source_path: str, file_id: str, file_extension: str
    ) -> str:
        """Copy a temp/upload source into upload_folder and return the destination path."""
        dest = self._persistent_upload_path(file_id, file_extension)
        shutil.copy2(source_path, dest)
        return dest

    def _file_path_from_db(
        self, file_id: str, agent_id: Optional[str] = None
    ) -> str:
        """
        Return file_path from file_sys_metadata.db ``files`` table.
        Falls back to the canonical upload_folder copy when legacy rows store temp paths.
        """
        row = self.db_service.get_file(file_id, agent_id=agent_id)
        if not row or len(row) < 5:
            return ""
        path = (row[4] or "").strip()
        if path and os.path.exists(path):
            return path
        file_type = (row[3] or "").strip().lower()
        if file_type:
            candidate = self._persistent_upload_path(file_id, file_type)
            if os.path.exists(candidate):
                return candidate
        upload_dir = Path(self.upload_folder)
        if upload_dir.is_dir():
            for match in sorted(upload_dir.glob(f"{file_id}.*")):
                if match.is_file():
                    return str(match)
        return path

    def _resolved_file_path_for_hit(self, hit: Dict[str, Any]) -> str:
        """Resolve file_path for a search/find hit from files.file_path in metadata DB."""
        file_id = hit.get("file_id") or hit.get("id")
        if not file_id:
            return (hit.get("file_path") or "").strip()
        agent_id = hit.get("agent_id")
        aid = str(agent_id).strip() if agent_id is not None and str(agent_id).strip() else None
        return self._file_path_from_db(file_id, aid) or (hit.get("file_path") or "").strip()

    def _enrich_hits_file_paths(self, hits: Optional[List[Dict[str, Any]]]) -> None:
        """In-place: set file_path on each hit from metadata DB (for @file / semantic search)."""
        for hit in hits or []:
            if isinstance(hit, dict) and (hit.get("file_id") or hit.get("id")):
                hit["file_path"] = self._resolved_file_path_for_hit(hit)

    def _apply_page_image_fields(
        self,
        hit: Dict[str, Any],
        file_id: str,
        agent_id: str,
        page_number: int,
    ) -> None:
        """Attach runtime page-image API URLs for frontend img tags."""
        apply_page_image_url_to_mapping(
            hit,
            file_id=str(file_id),
            agent_id=str(agent_id),
            page_number=int(page_number),
        )

    def _enrich_hits_preview_contract(self, hits: Optional[List[Dict[str, Any]]]) -> None:
        """In-place: preview_url, page_download_url, preview_mode for talkback UI."""
        for hit in hits or []:
            if not isinstance(hit, dict):
                continue
            file_id = hit.get("file_id") or hit.get("id")
            agent_id = hit.get("agent_id")
            if not file_id or agent_id is None or not str(agent_id).strip():
                continue
            aid = str(agent_id).strip()
            fid = str(file_id)
            apply_preview_contract_to_mapping(
                hit,
                file_id=fid,
                agent_id=aid,
                filename=str(hit.get("filename") or ""),
                file_type=str(hit.get("file_type") or ""),
                relative_url=True,
            )
            pn = hit.get("page_number")
            if pn is not None:
                try:
                    apply_page_download_url_to_mapping(
                        hit,
                        file_id=fid,
                        agent_id=aid,
                        page_number=int(pn),
                        relative_url=True,
                    )
                    self._apply_page_image_fields(hit, fid, aid, int(pn))
                except (TypeError, ValueError):
                    pass

    def _vector_index_document(
        self,
        file_id: str,
        content: str,
        doc_meta: dict,
        *,
        file_path: Optional[str] = None,
        file_ext: Optional[str] = None,
    ) -> int:
        """Index into Chroma + BM25 with one chunk per page."""
        if not self.vector_db:
            return 0

        page_texts: Optional[List[str]] = None
        ext = (file_ext or doc_meta.get("file_type") or "").lower()

        if file_path and ext in OFFICE_EXTENSIONS | {"pdf"}:
            _, page_texts, _ = prepare_ingestion(file_path, file_id, ext)

        ids, docs, metas = self.vector_db.chunk_document(
            file_id,
            content,
            doc_meta,
            page_texts=page_texts,
        )
        self.vector_db.add_chunks(ids, docs, metas)
        self._index_chunks_dense_and_sparse(ids, docs, metas)
        return len(ids)
    
    @property
    def db(self):
        """
        Backward compatibility property.
        Original DocumentAssistant used self.db for database operations.
        This alias allows routes to continue using doc_assistant.db.method_name()
        """
        return self.db_service 
        
    @property
    def _ingestion_agent_id(self) -> Optional[str]:
        """Return a single agent_id string for metadata tagging during ingestion."""
        if isinstance(self.agent_id, list):
            return self.agent_id[0] if self.agent_id else None
        return self.agent_id

    def _check_file_enabled(self) -> bool:
        return check_file_enabled_from_env_or_settings(settings)

    def _evaluate_check_file_upload(
        self,
        folder_path: str,
        relative_path: str,
        source_path: Optional[str] = None,
    ) -> Dict[str, Any]:
        return evaluate_check_file_upload(
            self.db_service,
            folder_path,
            relative_path,
            source_path=source_path,
            agent_id=self.agent_id,
        )

    def _replace_existing_file(self, file_id: str) -> bool:
        """Unregister prior duplicate (vectors, BM25, DB, derived assets); keep origin file_path."""
        file_info = self.db_service.get_file(file_id)
        source_path = ""
        if file_info and len(file_info) > 4:
            source_path = (file_info[4] or "").strip()
        ok = self.delete_document(file_id, preserve_source_path=True)
        if ok and source_path:
            print(
                f"[CHECK_FILE] Unregistered duplicate {file_id}; "
                f"preserved source file at {source_path}"
            )
        return ok

    @staticmethod
    def _preserve_disk_paths(file_info: tuple, preserve_source_path: bool) -> set:
        if not preserve_source_path or not file_info or len(file_info) <= 4:
            return set()
        path = (file_info[4] or "").strip()
        if not path:
            return set()
        return {os.path.normpath(os.path.abspath(path))}

    def _remove_disk_file_unless_preserved(self, path: str, preserve: set) -> None:
        if not path or not os.path.exists(path):
            return
        norm = os.path.normpath(os.path.abspath(path))
        if norm in preserve:
            return
        try:
            os.remove(path)
        except OSError as e:
            print(f"[FILE] Error removing file on disk {path}: {e}")

    def _apply_check_file_before_ingest(
        self,
        folder_path: str,
        relative_path: str,
        source_path: Optional[str] = None,
    ) -> Optional[Dict[str, Any]]:
        """
        When CHECK_FILE is enabled, return a skip dict if upload should not proceed,
        or None to continue (after replace cleanup if needed).
        """
        if not self._check_file_enabled():
            return None
        ev = self._evaluate_check_file_upload(folder_path, relative_path, source_path)
        print(
            f"[CHECK_FILE] folder={folder_path!r} rel={relative_path!r} "
            f"action={ev.get('action')} reason={ev.get('reason')} "
            f"search={ev.get('search_name')!r} replace_id={ev.get('existing_file_id')}"
        )
        action = ev.get("action")
        if action == "skip":
            reason = ev.get("reason") or "check_file_older"
            return {
                "uploaded": 0,
                "skipped": 1,
                "chunks": 0,
                "reason": reason,
                "success": False,
                "message": "Skipped: existing file is newer or same version",
            }
        if action == "replace" and ev.get("existing_file_id"):
            self._replace_existing_file(ev["existing_file_id"])
        return None
    
    def scan_folder_path(self, folder_path: str, max_file_size: Optional[int] = None, job_id: Optional[str] = None) -> Dict:
        """Scan folder and upload all files preserving structure.
        Skips files over max_file_size (default 100MB) to avoid memory issues.
        """
        if not os.path.exists(folder_path) or not os.path.isdir(folder_path):
            return {'success': False, 'message': 'Invalid folder path'}
        
        if max_file_size is None:
            max_file_size = self.DEFAULT_MAX_FILE_SIZE
        supported_extensions = ['pdf', 'docx', 'doc', 'xlsx', 'xls', 'csv', 'txt', 'png', 'jpg', 'jpeg']

        if job_id:
            from app.utils.file_schedule_utils import is_global_scheduler_resting
            from app.services.folder_scan_job_prompts import (
                FOLDER_SCAN_SCHEDULER_REST_SCAN_PAUSED,
            )

            if is_global_scheduler_resting():
                return {
                    "success": False,
                    "message": FOLDER_SCAN_SCHEDULER_REST_SCAN_PAUSED,
                    "uploaded": 0,
                    "skipped": 0,
                    "total_chunks": 0,
                    "rest_paused": True,
                    "cancelled": False,
                }

        uploaded_count = 0
        skipped_count = 0
        total_chunks = 0
        skip_reasons = []  # list of (filename, reason) for debugging

        root_folder_name = os.path.basename(os.path.abspath(folder_path))
        self.db_service.save_folder(root_folder_name, root_folder_name, "", folder_path)
        
        for root, dirs, files in os.walk(folder_path):
            rel_dir = os.path.relpath(root, folder_path)
            
            if rel_dir != '.':
                current_path_parts = rel_dir.split(os.sep)
                current_folder_path = root_folder_name
                for i, part in enumerate(current_path_parts):
                    parent_path = current_folder_path
                    current_folder_path = os.path.join(current_folder_path, part).replace(os.sep, '/')
                    full_sys_path = os.path.join(folder_path, *current_path_parts[:i+1])
                    self.db_service.save_folder(part, current_folder_path, parent_path, full_sys_path)
                folder_for_file = current_folder_path
            else:
                folder_for_file = root_folder_name
            
            for filename in files:
                if job_id:
                    from app.services.file_schedule_service import FileScheduleService
                    from app.utils.folder_scan_scheduler import is_folder_scan_cancel_requested

                    sched_svc = FileScheduleService()
                    if is_folder_scan_cancel_requested(job_id):
                        print(f"Job {job_id} was cancelled mid-run. Exiting scan_folder_path early.")
                        return {
                            'success': False,
                            'message': 'Job cancelled by user',
                            'uploaded': uploaded_count,
                            'skipped': skipped_count,
                            'total_chunks': total_chunks,
                            'cancelled': True
                        }
                    
                    from app.utils.file_schedule_utils import is_global_scheduler_resting
                    from app.services.folder_scan_job_prompts import (
                        FOLDER_SCAN_SCHEDULER_REST_SCAN_PAUSED,
                    )

                    if is_global_scheduler_resting():
                        print(
                            f"Job {job_id} paused — scheduler entered rest phase. "
                            "Exiting scan_folder_path without ingesting files.",
                            flush=True,
                        )
                        return {
                            "success": False,
                            "message": FOLDER_SCAN_SCHEDULER_REST_SCAN_PAUSED,
                            "uploaded": uploaded_count,
                            "skipped": skipped_count,
                            "total_chunks": total_chunks,
                            "rest_paused": True,
                            "cancelled": False,
                        }

                    status = sched_svc.get_scheduler_status()
                    if status:
                        phase = (status.get("phase") or "active").lower()
                        next_phase_at_str = status.get("next_phase_at")
                        if next_phase_at_str and phase == "active":
                            next_phase_at = datetime.fromisoformat(next_phase_at_str)
                            remaining_secs = (
                                next_phase_at - datetime.now()
                            ).total_seconds()
                            from app.utils.folder_scan_scheduler import (
                                log_remaining_minutes_if_needed,
                            )

                            log_remaining_minutes_if_needed(phase, remaining_secs)
                # Skip macOS AppleDouble/resource-fork files (._*); they are not real documents
                if filename.startswith('._'):
                    skipped_count += 1
                    skip_reasons.append((filename, "resource_fork_metadata"))
                    continue
                file_ext = filename.split('.')[-1].lower()
                if file_ext not in supported_extensions:
                    skipped_count += 1
                    skip_reasons.append((filename, "unsupported_extension"))
                    continue
                
                file_full_path = os.path.join(root, filename)
                relative_file_path = os.path.relpath(file_full_path, folder_path)
                
                try:
                    if self._check_file_enabled():
                        skip = self._apply_check_file_before_ingest(
                            folder_for_file,
                            relative_file_path,
                            source_path=file_full_path,
                        )
                        if skip:
                            skipped_count += 1
                            skip_reasons.append((filename, skip.get("reason", "check_file_older")))
                            continue
                    else:
                        skip = self._skip_if_file_already_ingested(file_full_path, filename)
                        if skip:
                            skipped_count += 1
                            skip_reasons.append(
                                (filename, skip.get("reason", "file_already_exists"))
                            )
                            continue

                    file_size = os.path.getsize(file_full_path)
                    if file_size > max_file_size:
                        skipped_count += 1
                        skip_reasons.append((filename, f"file_too_large ({file_size} bytes, max {max_file_size})"))
                        print(f"Skipped (size {file_size} > {max_file_size}): {filename}")
                        continue
                    
                    content, _ = self.process_file_from_path(file_full_path, file_ext)
                    
                    if content and not content.startswith("Error") and not content.startswith("Unsupported"):
                        file_id = self.get_file_hash(content, filename, relative_file_path)
                        if self.db_service.get_file(file_id):
                            skipped_count += 1
                            skip_reasons.append((filename, "duplicate"))
                            continue
                        
                        saved_path = file_full_path
                        image_path, thumbnail_path = "", ""
                        self.db_service.save_file(
                            file_id, filename, file_ext, saved_path,
                            content, folder_for_file, relative_file_path,
                            image_path or "", thumbnail_path or "", self.agent_id
                        )
                        if self.vector_db:
                            doc_meta = {
                                "filename": filename,
                                "file_type": file_ext,
                                "folder_path": folder_for_file,
                                "relative_path": relative_file_path,
                                "upload_date": datetime.now().isoformat(),
                                "agent_id": self._ingestion_agent_id,
                            }
                            chunks = self._vector_index_document(
                                file_id,
                                content,
                                doc_meta,
                                file_path=saved_path,
                                file_ext=file_ext,
                            )
                            total_chunks += chunks
                        uploaded_count += 1
                    else:
                        skipped_count += 1
                        reason = (content or "empty")[:80] if content else "empty_content"
                        skip_reasons.append((filename, f"content_error: {reason}"))
                        print(f"Skipped (content): {filename} - {reason}")
                except Exception as e:
                    skipped_count += 1
                    skip_reasons.append((filename, f"exception: {str(e)}"))
                    print(f"Error processing {filename}: {e}")
        
        return {
            'success': True,
            'uploaded': uploaded_count,
            'skipped': skipped_count,
            'total_chunks': total_chunks,
            'folder_name': root_folder_name,
            'skip_reasons': skip_reasons[:50],
        }
    
    def _get_all_files_in_folder_recursive(self, folder_path: str) -> List[tuple]:
        """Get all files in folder and all its subfolders recursively"""
        all_files = []
        
        files = self.db_service.get_files_in_folder(folder_path)
        all_files.extend(files)
        
        all_folders = self.db_service.get_all_folders()
        
        for folder_info in all_folders:
            folder_name, current_path, file_count, parent_path, full_path = folder_info
            
            if current_path.startswith(folder_path + '/'):
                subfolder_files = self.db_service.get_files_in_folder(current_path)
                all_files.extend(subfolder_files)
        
        return all_files
    
    # Max file size for scan_folder_path (bytes); larger files are skipped to avoid memory issues
    DEFAULT_MAX_FILE_SIZE = 100 * 1024 * 1024  # 100 MB

    SCAN_SUPPORTED_EXTENSIONS = (
        "pdf",
        "docx",
        "doc",
        "xlsx",
        "xls",
        "csv",
        "txt",
        "png",
        "jpg",
        "jpeg",
    )

    @staticmethod
    def _build_dirs_dfs_preorder(scan_root_abs: str) -> List[str]:
        """DFS preorder list of relative directory paths from scan root ('' = root)."""
        out: List[str] = []

        def visit(rel_norm: str) -> None:
            out.append(rel_norm)
            full = (
                os.path.join(scan_root_abs, *rel_norm.split("/"))
                if rel_norm
                else scan_root_abs
            )
            if not os.path.isdir(full):
                return
            try:
                subs = sorted(
                    n for n in os.listdir(full) if os.path.isdir(os.path.join(full, n))
                )
            except OSError:
                return
            for name in subs:
                sub = f"{rel_norm}/{name}" if rel_norm else name
                visit(sub)

        visit("")
        return out

    def _ensure_folders_for_scan_rel(
        self, scan_root_abs: str, rel_from_scan: str, root_folder_name: str
    ) -> str:
        """Ensure folder rows exist for rel_from_scan; return logical folder_path for file rows."""
        rel_norm = (rel_from_scan or "").replace("\\", "/").strip("/")
        if not rel_norm:
            self.db_service.save_folder(root_folder_name, root_folder_name, "", scan_root_abs)
            return root_folder_name
        parts = rel_norm.split("/")
        current_folder_path = root_folder_name
        for i, part in enumerate(parts):
            parent_path = current_folder_path
            current_folder_path = "/".join([root_folder_name, *parts[: i + 1]])
            full_sys_path = os.path.join(scan_root_abs, *parts[: i + 1])
            self.db_service.save_folder(part, current_folder_path, parent_path, full_sys_path)
        return current_folder_path

    def _list_scan_candidate_filenames(self, scan_root_abs: str, rel_from_scan: str) -> List[str]:
        rel_norm = (rel_from_scan or "").replace("\\", "/").strip("/")
        full = (
            os.path.join(scan_root_abs, *rel_norm.split("/"))
            if rel_norm
            else scan_root_abs
        )
        if not os.path.isdir(full):
            return []
        names: List[str] = []
        try:
            for name in sorted(os.listdir(full)):
                if name.startswith("._"):
                    continue
                fp = os.path.join(full, name)
                if not os.path.isfile(fp):
                    continue
                ext = name.split(".")[-1].lower()
                if ext not in DocumentService.SCAN_SUPPORTED_EXTENSIONS:
                    continue
                names.append(name)
        except OSError:
            return []
        return names

    @staticmethod
    def _logical_folder_for_scan_rel(root_folder_name: str, rel_dir: str) -> str:
        """Logical folder_path for a directory under a scan root (no DB writes)."""
        rel_norm = (rel_dir or "").replace("\\", "/").strip("/")
        if rel_norm in ("", "."):
            return root_folder_name
        return f"{root_folder_name}/{rel_norm}".replace("\\", "/")

    def _skip_if_file_already_ingested(
        self, disk_file_path: str, file_name: str
    ) -> Optional[Dict[str, Any]]:
        """Per-file dedupe by disk ``file_path`` + file name (not whole ``folder_path``)."""
        if self._check_file_enabled():
            return None
        disk_norm = os.path.normpath(os.path.abspath(str(disk_file_path or "").strip()))
        name = str(file_name or "").strip()
        if self.db_service.get_file_by_disk_path_and_name(disk_norm, name):
            return {
                "uploaded": 0,
                "skipped": 1,
                "chunks": 0,
                "reason": "file_already_exists",
            }
        return None

    def _ingest_single_scan_file(
        self,
        scan_root_abs: str,
        filename: str,
        rel_dir: str,
        folder_for_file: str,
        max_file_size: int,
        job_id: Optional[str] = None,
    ) -> Dict[str, Union[int, str]]:
        rel_dir_norm = (rel_dir or "").replace("\\", "/").strip("/")
        parts = rel_dir_norm.split("/") if rel_dir_norm else []
        file_full_path = os.path.join(scan_root_abs, *parts, filename)
        if not os.path.isfile(file_full_path):
            return {"uploaded": 0, "skipped": 1, "chunks": 0, "reason": "not_a_file"}
        relative_file_path = os.path.relpath(file_full_path, scan_root_abs).replace(os.sep, "/")
        file_ext = filename.split(".")[-1].lower()
        ingest_started = False

        was_active_before = False
        from app.services.file_schedule_service import FileScheduleService
        from app.utils.file_schedule_utils import seconds_until_phase_end
        sched_svc = FileScheduleService()
        if job_id:
            sched = sched_svc.get_file_schedule(file_path=scan_root_abs, agent_id=self.agent_id)
            if sched and int(sched.get("schedule_enabled") or 0):
                if (sched.get("phase") or "active").lower() == "active":
                    was_active_before = seconds_until_phase_end(sched) > 0

        try:
            if job_id:
                from app.utils.folder_scan_scheduler import (
                    folder_scan_ingest_begin,
                    folder_scan_ingest_end,
                    is_folder_scan_cancel_requested,
                )

                if is_folder_scan_cancel_requested(job_id):
                    return {
                        "uploaded": 0,
                        "skipped": 0,
                        "chunks": 0,
                        "reason": "cancelled",
                        "cancelled": True,
                    }
                folder_scan_ingest_begin(job_id)
                ingest_started = True
            if self._check_file_enabled():
                skip = self._apply_check_file_before_ingest(
                    folder_for_file,
                    relative_file_path,
                    source_path=file_full_path,
                )
                if skip:
                    return skip
            else:
                skip = self._skip_if_file_already_ingested(file_full_path, filename)
                if skip:
                    return skip

            file_size = os.path.getsize(file_full_path)
            if file_size > max_file_size:
                return {"uploaded": 0, "skipped": 1, "chunks": 0, "reason": "file_too_large"}

            content, _ = self.process_file_from_path(
                file_full_path, file_ext, job_id=job_id
            )
            if content == FOLDER_SCAN_JOB_STOPPED:
                return {
                    "uploaded": 0,
                    "skipped": 0,
                    "chunks": 0,
                    "reason": "cancelled",
                    "cancelled": True,
                }
            if not content or content.startswith("Error") or content.startswith("Unsupported"):
                return {"uploaded": 0, "skipped": 1, "chunks": 0, "reason": "content_error"}
            file_id = self.get_file_hash(content, filename, relative_file_path)
            if self.db_service.get_file(file_id):
                return {"uploaded": 0, "skipped": 1, "chunks": 0, "reason": "duplicate"}
            saved_path = file_full_path
            image_path, thumbnail_path = "", ""
            self.db_service.save_file(
                file_id,
                filename,
                file_ext,
                saved_path,
                content,
                folder_for_file,
                relative_file_path,
                image_path or "",
                thumbnail_path or "",
                self.agent_id,
            )
            out_chunks = 0
            if self.vector_db:
                doc_meta = {
                    "filename": filename,
                    "file_type": file_ext,
                    "folder_path": folder_for_file,
                    "relative_path": relative_file_path,
                    "upload_date": datetime.now().isoformat(),
                    "agent_id": self._ingestion_agent_id, # Tag chunks with agent_id for isolation
                }
                out_chunks = self._vector_index_document(
                    file_id,
                    content,
                    doc_meta,
                    file_path=saved_path,
                    file_ext=file_ext,
                )
            return {"uploaded": 1, "skipped": 0, "chunks": out_chunks, "reason": "ok"}
        except Exception as e:
            print(f"Error processing {filename}: {e}")
            return {"uploaded": 0, "skipped": 1, "chunks": 0, "reason": str(e)[:200]}
        finally:
            if was_active_before and job_id:
                sched_after = sched_svc.get_file_schedule(file_path=scan_root_abs, agent_id=self.agent_id)
                if sched_after and int(sched_after.get("schedule_enabled") or 0):
                    if (sched_after.get("phase") or "active").lower() == "active":
                        if seconds_until_phase_end(sched_after) <= 0:
                            from app.utils.file_schedule_utils import (
                                remaining_rest_after_upload_overflow,
                            )

                            remaining_rest, _, _ = remaining_rest_after_upload_overflow()
                            sched_svc.update_file_schedule_phase(
                                file_path=scan_root_abs,
                                agent_id=self.agent_id,
                                phase="rest",
                                override_dur_seconds=max(1, remaining_rest),
                            )
            if ingest_started and job_id:
                from app.utils.folder_scan_scheduler import folder_scan_ingest_end

                folder_scan_ingest_end(job_id)

    def start_resumable_folder_scan(
        self,
        folder_path: str,
        batch_size: int = 100,
        max_file_size: Optional[int] = None,
    ) -> Dict:
        """Create a resumable folder scan job. Call process_folder_scan_job_batch until completed."""
        scan_root = os.path.abspath(folder_path)
        if not os.path.isdir(scan_root):
            return {"success": False, "message": FOLDER_SCAN_JOB_INVALID_PATH}
        if max_file_size is None:
            max_file_size = self.DEFAULT_MAX_FILE_SIZE
        batch_size = max(1, min(int(batch_size or 100), 500))
        root_folder_name = os.path.basename(scan_root)
        dirs_order = self._build_dirs_dfs_preorder(scan_root)
        if not dirs_order:
            return {"success": False, "message": FOLDER_SCAN_JOB_INVALID_PATH}
        self._ensure_folders_for_scan_rel(scan_root, "", root_folder_name)
        job_id = str(uuid.uuid4())
        if not self.db_service.create_folder_scan_job(
            job_id=job_id,
            root_path=scan_root,
            root_folder_name=root_folder_name,
            dirs_json=json.dumps(dirs_order, ensure_ascii=False),
            batch_size=batch_size,
            max_file_size=max_file_size,
        ):
            return {
                "success": False,
                "message": FOLDER_SCAN_JOB_FAILED,
                "details": {"reason": "db_insert_failed"},
            }
        return {
            "success": True,
            "message": FOLDER_SCAN_JOB_STARTED,
            "job_id": job_id,
            "root_path": scan_root,
            "root_folder_name": root_folder_name,
            "dirs_total": len(dirs_order),
            "batch_size": batch_size,
            "max_file_size": max_file_size,
            "completed": False,
        }

    def process_folder_scan_job_batch(self, job_id: str, batch_size: Optional[int] = None) -> Dict:
        """Process up to ``batch_size`` file ingest attempts this call, moving across DFS dirs as needed."""
        row = self.db_service.get_folder_scan_job(job_id)
        if not row:
            return {"success": False, "message": FOLDER_SCAN_JOB_NOT_FOUND, "completed": True}
        if row.get("status") == "completed":
            stats = json.loads(row.get("stats_json") or "{}")
            return {
                "success": True,
                "message": FOLDER_SCAN_JOB_COMPLETED,
                "job_id": job_id,
                "completed": True,
                "uploaded_batch": 0,
                "skipped_batch": 0,
                "chunks_batch": 0,
                **stats,
            }
        if row.get("status") == "failed":
            return {
                "success": False,
                "message": FOLDER_SCAN_JOB_FAILED,
                "job_id": job_id,
                "completed": True,
                "last_error": row.get("last_error"),
            }
        scan_root = str(row["root_path"])
        root_folder_name = str(row["root_folder_name"])
        dirs_order = json.loads(row["dirs_json"] or "[]")
        if not isinstance(dirs_order, list) or not dirs_order:
            self.db_service.update_folder_scan_job(job_id, status="failed", last_error="Invalid dirs_json")
            return {"success": False, "message": FOLDER_SCAN_JOB_FAILED, "completed": True}
        bs = int(batch_size or row.get("batch_size") or 100)
        bs = max(1, min(bs, 500))
        max_file_size = int(row["max_file_size"] or self.DEFAULT_MAX_FILE_SIZE)
        dir_index = int(row["dir_index"] or 0)
        file_offset = int(row["file_offset"] or 0)
        stats = json.loads(row.get("stats_json") or "{}")
        tot_u = int(stats.get("uploaded", 0))
        tot_s = int(stats.get("skipped", 0))
        tot_c = int(stats.get("total_chunks", 0))
        bu = sk = ch = 0
        current_rel = ""
        try:
            self.db_service.update_folder_scan_job(job_id, status="running")
            remaining = bs
            while True:
                if remaining <= 0:
                    break
                if dir_index >= len(dirs_order):
                    break
                current_rel = str(dirs_order[dir_index] or "")
                folder_for_file = self._ensure_folders_for_scan_rel(
                    scan_root, current_rel, root_folder_name
                )
                names = self._list_scan_candidate_filenames(scan_root, current_rel)
                if file_offset >= len(names):
                    dir_index += 1
                    file_offset = 0
                    continue
                take = min(remaining, len(names) - file_offset)
                chunk = names[file_offset : file_offset + take]
                for fn in chunk:
                    try:
                        from app.utils.folder_scan_scheduler import (
                            is_folder_scan_cancel_requested,
                        )

                        if is_folder_scan_cancel_requested(job_id):
                            stats_json = json.dumps(
                                {"uploaded": tot_u, "skipped": tot_s, "total_chunks": tot_c}
                            )
                            self.db_service.update_folder_scan_job(
                                job_id,
                                dir_index=dir_index,
                                file_offset=file_offset,
                                status="cancelled",
                                stats_json=stats_json,
                                last_error="Upload schedule stopped by user",
                            )
                            return {
                                "success": True,
                                "message": FOLDER_SCAN_JOB_STOPPED,
                                "job_id": job_id,
                                "completed": False,
                                "cancelled": True,
                                "dir_index": dir_index,
                                "file_offset": file_offset,
                                "dirs_total": len(dirs_order),
                                "current_rel": current_rel,
                                "uploaded_batch": bu,
                                "skipped_batch": sk,
                                "chunks_batch": ch,
                                "uploaded": tot_u,
                                "skipped": tot_s,
                                "total_chunks": tot_c,
                            }

                        from app.services.file_schedule_service import FileScheduleService
                        from app.utils.file_schedule_utils import seconds_until_phase_end

                        sched_svc = FileScheduleService()
                        sched_row = sched_svc.get_file_schedule(
                            file_path=scan_root, agent_id=self.agent_id
                        )
                        if sched_row and int(sched_row.get("schedule_enabled") or 0):
                            phase = (sched_row.get("phase") or "active").lower()
                            if phase == "rest" or (phase == "active" and seconds_until_phase_end(sched_row) <= 0):
                                if phase == "active":
                                    sched_svc.update_file_schedule_phase(
                                        file_path=scan_root,
                                        agent_id=self.agent_id,
                                        phase="rest",
                                    )
                                break
                    except Exception:
                        pass
                    r = self._ingest_single_scan_file(
                        scan_root,
                        fn,
                        current_rel,
                        folder_for_file,
                        max_file_size,
                        job_id=job_id,
                    )
                    if r.get("cancelled"):
                        stats_json = json.dumps(
                            {"uploaded": tot_u, "skipped": tot_s, "total_chunks": tot_c}
                        )
                        self.db_service.update_folder_scan_job(
                            job_id,
                            dir_index=dir_index,
                            file_offset=file_offset,
                            status="cancelled",
                            stats_json=stats_json,
                            last_error="Upload schedule stopped by user",
                        )
                        return {
                            "success": True,
                            "message": FOLDER_SCAN_JOB_STOPPED,
                            "job_id": job_id,
                            "completed": False,
                            "cancelled": True,
                            "dir_index": dir_index,
                            "file_offset": file_offset,
                            "dirs_total": len(dirs_order),
                            "current_rel": current_rel,
                            "uploaded_batch": bu,
                            "skipped_batch": sk,
                            "chunks_batch": ch,
                            "uploaded": tot_u,
                            "skipped": tot_s,
                            "total_chunks": tot_c,
                        }
                    du = int(r.get("uploaded", 0))
                    ds = int(r.get("skipped", 0))
                    dc = int(r.get("chunks", 0))
                    bu += du
                    sk += ds
                    ch += dc
                    tot_u += du
                    tot_s += ds
                    tot_c += dc
                    if self.agent_id:
                        try:
                            from app.utils.folder_scan_scheduler import (
                                announce_scheduler_remaining_on_file_upload,
                            )

                            announce_scheduler_remaining_on_file_upload(
                                scan_root, str(self.agent_id), fn
                            )
                        except Exception:
                            pass
                file_offset += len(chunk)
                remaining -= len(chunk)
                if file_offset >= len(names):
                    dir_index += 1
                    file_offset = 0
            done_all = dir_index >= len(dirs_order)
            stats_json = json.dumps({"uploaded": tot_u, "skipped": tot_s, "total_chunks": tot_c})
            self.db_service.update_folder_scan_job(
                job_id,
                dir_index=dir_index,
                file_offset=file_offset,
                status="completed" if done_all else "running",
                stats_json=stats_json,
            )
            return {
                "success": True,
                "message": FOLDER_SCAN_JOB_COMPLETED if done_all else FOLDER_SCAN_JOB_STEP_PROGRESS,
                "job_id": job_id,
                "completed": bool(done_all),
                "dir_index": dir_index,
                "file_offset": file_offset,
                "dirs_total": len(dirs_order),
                "current_rel": current_rel,
                "uploaded_batch": bu,
                "skipped_batch": sk,
                "chunks_batch": ch,
                "uploaded": tot_u,
                "skipped": tot_s,
                "total_chunks": tot_c,
            }
        except Exception as e:
            err = str(e)[:2000]
            self.db_service.update_folder_scan_job(
                job_id,
                status="failed",
                last_error=err,
                stats_json=json.dumps({"uploaded": tot_u, "skipped": tot_s, "total_chunks": tot_c}),
            )
            return {
                "success": False,
                "message": FOLDER_SCAN_JOB_FAILED,
                "completed": True,
                "last_error": err,
            }

    def get_file_hash(self, content: str, filename: str, relative_path: str = "", agent_id: str = None) -> str:
        """Generate unique file hash from path, content, and agent_id to allow same file for different agents."""
        path_part = f"{filename}{relative_path}"
        content_sample = content[:2000] if len(content) > 2000 else content
        size_part = str(len(content))
        target_agent_id = agent_id if agent_id is not None else getattr(self, "agent_id", "")
        agent_part = str(target_agent_id)
        return hashlib.md5(f"{path_part}{size_part}{agent_part}{content_sample}".encode()).hexdigest()
    
    def process_file(self, uploaded_file, file_type: str) -> Tuple[str, str]:
        """Process uploaded file and extract content"""
        print(f"[FILE] process_file start: type={file_type}")
        if hasattr(uploaded_file, 'seek'):
            uploaded_file.seek(0)
        
        extractors = {
            'pdf': self.content_extractor.extract_from_pdf,
            'xlsx': self.content_extractor.extract_from_excel,
            'xls': self.content_extractor.extract_from_excel,
            'csv': self.content_extractor.extract_from_csv,
            'png': self.content_extractor.extract_from_image,
            'jpg': self.content_extractor.extract_from_image,
            'jpeg': self.content_extractor.extract_from_image,
            'docx': self.content_extractor.extract_from_docx,
            'doc': self.content_extractor.extract_from_doc,
            'txt': self.content_extractor.extract_from_txt
        }
        
        extractor = extractors.get(file_type)
        if extractor:
            print(f"[FILE] process_file extractor selected: {extractor.__name__}")
            content = extractor(uploaded_file)
            print(f"[FILE] process_file done: chars={len(content) if isinstance(content, str) else 'n/a'}")
            return content, file_type
        print(f"[FILE] process_file unsupported type: {file_type}")
        return "Unsupported file type", None

    def process_file_from_path(
        self,
        file_full_path: str,
        file_type: str,
        job_id: Optional[str] = None,
    ) -> Tuple[str, str]:
        """Process file from disk path. Uses path for PDF/Office to avoid loading large files into memory."""
        cancel_check = None
        if job_id:
            from app.utils.folder_scan_scheduler import is_folder_scan_cancel_requested

            cancel_check = lambda: is_folder_scan_cancel_requested(job_id)
            if cancel_check():
                return FOLDER_SCAN_JOB_STOPPED, file_type
        if file_type == 'pdf':
            content = self._content_from_saved_path(
                file_full_path, file_type, should_cancel=cancel_check
            )
            if cancel_check and cancel_check():
                return FOLDER_SCAN_JOB_STOPPED, file_type
            return content, file_type
        if file_type in OFFICE_EXTENSIONS:
            content = self._content_from_saved_path(file_full_path, file_type)
            return content, file_type
        with open(file_full_path, 'rb') as f:
            file_bytes = io.BytesIO(f.read())
            file_bytes.name = os.path.basename(file_full_path)
        return self.process_file(file_bytes, file_type)
    
    def upload_document(self, uploaded_file, folder_path: str = "") -> Dict:
        """Upload a single document"""
        print(f"[FILE] upload_document start: name={getattr(uploaded_file, 'filename', '')}, folder={folder_path}")
        if not self.vector_db:
            return {'success': False, 'message': 'OpenAI API key not set'}
        
        file_type = uploaded_file.filename.split('.')[-1].lower()
        rel_path = (uploaded_file.filename or "").replace("\\", "/")

        if self._check_file_enabled():
            skip = self._apply_check_file_before_ingest(folder_path, rel_path)
            if skip:
                return {
                    "success": False,
                    "message": skip.get("message", "Skipped: existing file is newer"),
                    "reason": skip.get("reason"),
                }
        elif self.db_service.get_file_by_path(folder_path, uploaded_file.filename):
            return {'success': False, 'message': 'File path already exists'}
        
        if hasattr(uploaded_file, 'seek'):
            uploaded_file.seek(0)
        raw_bytes = uploaded_file.read()
        rel_name = os.path.basename((uploaded_file.filename or "file").replace("\\", "/"))
        logical = (folder_path or "").strip().replace("\\", "/").strip("/")
        file_path = os.path.normpath(os.path.join(
            os.path.abspath(self.upload_folder),
            *([logical, rel_name] if logical else [rel_name]),
        ))
        os.makedirs(os.path.dirname(file_path), exist_ok=True)
        with open(file_path, "wb") as f:
            f.write(raw_bytes)

        if file_type in OFFICE_EXTENSIONS | {"pdf"}:
            content = self._content_from_saved_path(file_path, file_type)
        else:
            content, _ = self.process_file(io.BytesIO(raw_bytes), file_type)
        
        if not content or content.startswith("Error") or content.startswith("Unsupported"):
            try:
                os.remove(file_path)
            except OSError:
                pass
            return {'success': False, 'message': content}
        
        file_id = self.get_file_hash(content, uploaded_file.filename)
        print(f"[FILE] upload_document hash computed: file_id={file_id}")
        
        if self.db_service.get_file(file_id):
            try:
                os.remove(file_path)
            except OSError:
                pass
            return {'success': False, 'message': 'File already exists'}
        
        image_path, thumbnail_path = "", ""
        
        self.db_service.save_file(
            file_id, uploaded_file.filename, file_type, file_path,
            content, folder_path, uploaded_file.filename, image_path or "", thumbnail_path or "", self.agent_id
        )
        print(f"[FILE] upload_document metadata saved: file_id={file_id}")
        
        metadata = {
            "filename": uploaded_file.filename,
            "file_type": file_type,
            "folder_path": folder_path,
            "upload_date": datetime.now().isoformat(),
            "agent_id": self._ingestion_agent_id
        }
        
        chunks_created = self._vector_index_document(
            file_id,
            content,
            metadata,
            file_path=file_path,
            file_ext=file_type,
        )
        print(f"[FILE] upload_document indexed: file_id={file_id}, chunks={chunks_created}")
        
        return {
            'success': True,
            'message': 'Upload successful',
            'file_id': file_id,
            'chunks_created': chunks_created,
            'images_extracted': 0,
            'folder_path': folder_path
        }
    
    def upload_document_from_bytes(self, file_bytes: bytes, filename: str, folder_path: str = "") -> Dict:
        """
        Upload a document from raw bytes (for base64 encoded uploads).
        
        Args:
            file_bytes: Raw file content as bytes
            filename: Name of the file with extension
            folder_path: Destination folder path
            
        Returns:
            Dict with success status and file info
        """
        print(f"[FILE] upload_document_from_bytes start: name={filename}, folder={folder_path}, bytes={len(file_bytes)}")
        if not self.vector_db:
            return {'success': False, 'message': 'OpenAI API key not set'}
        
        # Create BytesIO object from bytes (non-PDF types only)
        file_type = filename.split('.')[-1].lower()
        rel_path = (filename or "").replace("\\", "/")

        if self._check_file_enabled():
            skip = self._apply_check_file_before_ingest(folder_path, rel_path)
            if skip:
                return {
                    "success": False,
                    "message": skip.get("message", "Skipped: existing file is newer"),
                    "reason": skip.get("reason"),
                }
        elif self.db_service.get_file_by_path(folder_path, filename):
            return {'success': False, 'message': 'File path already exists'}

        rel_name = os.path.basename((filename or "file").replace("\\", "/"))
        logical = (folder_path or "").strip().replace("\\", "/").strip("/")
        file_path = os.path.normpath(os.path.join(
            os.path.abspath(self.upload_folder),
            *([logical, rel_name] if logical else [rel_name]),
        ))
        os.makedirs(os.path.dirname(file_path), exist_ok=True)
        with open(file_path, "wb") as f:
            f.write(file_bytes)

        if file_type in OFFICE_EXTENSIONS | {"pdf"}:
            content = self._content_from_saved_path(file_path, file_type)
        else:
            content, _ = self.process_file(io.BytesIO(file_bytes), file_type)
        
        if not content or content.startswith("Error") or content.startswith("Unsupported"):
            try:
                os.remove(file_path)
            except OSError:
                pass
            return {'success': False, 'message': content or 'Failed to extract content'}
        
        file_id = self.get_file_hash(content, filename)
        print(f"[FILE] upload_document_from_bytes hash computed: file_id={file_id}")
        
        if self.db_service.get_file(file_id):
            try:
                os.remove(file_path)
            except OSError:
                pass
            return {'success': False, 'message': 'File already exists'}
        
        image_path, thumbnail_path = "", ""
        
        self.db_service.save_file(
            file_id, filename, file_type, file_path,
            content, folder_path, filename, image_path or "", thumbnail_path or "", self.agent_id
        )
        print(f"[FILE] upload_document_from_bytes metadata saved: file_id={file_id}")
        
        metadata = {
            "filename": filename,
            "file_type": file_type,
            "folder_path": folder_path,
            "upload_date": datetime.now().isoformat(),
            "agent_id": self._ingestion_agent_id # Metadata tag for agent isolation
        }
        
        chunks_created = self._vector_index_document(
            file_id,
            content,
            metadata,
            file_path=file_path,
            file_ext=file_type,
        )
        print(f"[FILE] upload_document_from_bytes indexed: file_id={file_id}, chunks={chunks_created}")
        
        return {
            'success': True,
            'message': 'Upload successful',
            'file_id': file_id,
            'chunks_created': chunks_created,
            'images_extracted': 0,
            'folder_path': folder_path
        }
    
    def _detect_folder_from_query(self, query: str) -> Optional[str]:
        """Detect folder from query using lightweight regex/substring matching."""
        folders = self.db_service.get_all_folders()
        if not folders:
            return None

        available_folders = [folder[1] for folder in folders if folder and len(folder) > 1 and folder[1]]
        if not available_folders:
            return None

        query_l = (query or "").strip().lower()
        if not query_l:
            return None

        # Pattern examples:
        # - "in reports"
        # - "from ml folder"
        # - "inside finance"
        m = re.search(r"\b(?:in|from|inside|under)\s+([a-z0-9_\-/ ]{2,80})", query_l)
        candidate = m.group(1).strip() if m else None
        if candidate:
            candidate = re.sub(r"\bfolder\b", "", candidate).strip(" .,_-/")
            if candidate:
                for folder in available_folders:
                    f = folder.lower()
                    if candidate == f or candidate in f or f in candidate:
                        return folder

        for folder in sorted(available_folders, key=lambda p: len(p or ""), reverse=True):
            f = (folder or "").strip().lower()
            if not f:
                continue
            if len(f) <= 2:
                if re.search(rf"(?<![a-z0-9_/]){re.escape(f)}(?![a-z0-9_])", query_l):
                    return folder
                continue
            if f in query_l:
                return folder

        return None

    def _format_chunk_hit(
        self,
        ch: Any,
        relevance: float,
        file_info: tuple,
    ) -> Dict[str, Any]:
        """Stable chunk-level hit for search / find responses."""
        meta = (ch.metadata if hasattr(ch, "metadata") else {}) or {}
        file_id = meta.get("file_id") or file_info[0]
        text = (ch.text if hasattr(ch, "text") else "") or ""
        hit: Dict[str, Any] = {
            "chunk_id": ch.chunk_id if hasattr(ch, "chunk_id") else meta.get("chunk_id", ""),
            "file_id": file_id,
            "id": file_id,
            "filename": file_info[2],
            "file_type": file_info[3],
            "file_path": "",
            "folder_path": file_info[5],
            "relative_path": file_info[6],
            "upload_date": file_info[7],
            "relevance": round(float(relevance), 3),
            "chunk_text": text,
            "content": text,
            "chunk_index": meta.get("chunk_index", 0),
            "total_chunks": meta.get("total_chunks", 1),
            "image_path": file_info[11] if len(file_info) > 11 else None,
            "thumbnail_path": file_info[12] if len(file_info) > 12 else None,
            "agent_id": file_info[13] if len(file_info) > 13 else meta.get("agent_id"),
            "distance": getattr(ch, "distance", None),
            "rerank_score": getattr(ch, "rerank_score", None),
            "bm25_score": getattr(ch, "bm25_score", None),
        }
        page_number = page_number_from_metadata(meta)
        if page_number is not None:
            hit["page_number"] = page_number
        hit["file_path"] = self._resolved_file_path_for_hit(hit) or (
            self._file_path_from_db(
                file_id,
                str(file_info[13]) if len(file_info) > 13 and file_info[13] else None,
            )
            or file_info[4]
        )
        self._enrich_hits_preview_contract([hit])
        return hit

    def search_semantic(
        self,
        query: str,
        folder_path: str = None,
        n_results: int = 10,
        trace=None,
    ) -> Dict[str, List[Dict]]:
        """
        Semantic search returning chunk-ranked hits.

        Returns:
            {"chunk_hits": [...], "results": [...]} where results is a legacy
            subset (first n_results) of the same chunk rows.
        """
        empty = {"chunk_hits": [], "results": []}
        print(f"📁🔎 [FILE][semantic] start query='{query[:80]}' folder='{folder_path}' n_results={n_results}")
        if not self.vector_db:
            return empty

        if not folder_path:
            folder_path = self._detect_folder_from_query(query)
        n = max(1, int(n_results or 10))
        probe = min(max(n * 6, 32), 150)
        cfg = DocumentRAGConfig(
            dense_k=probe,
            sparse_k=probe,
            fused_k=20,
            rerank_k=10,
            compress_keep_ratio=1.0,
        )
        pipeline = DocumentRAGPipeline(self, config=cfg)
        state = pipeline.run(query=query, folder_path=folder_path, trace=trace)
        chunks = (state.context.chunks if state.context else []) or []
        print(
            f"📁🧠 [FILE][semantic] pipeline done chunks={len(chunks)} errors={len(state.errors)}"
        )

        relevances = chunk_relevance_scores_0_100(chunks, query=query)
        ranked: List[Tuple[Any, float]] = sorted(
            zip(chunks, relevances), key=lambda x: x[1], reverse=True
        )

        chunk_hits: List[Dict] = []
        for ch, relevance in ranked[:TOP_CHUNK_HITS]:
            meta = ch.metadata or {}
            file_id = meta.get("file_id")
            if not file_id:
                continue
            agent_id = meta.get("agent_id")
            file_info = self.db_service.get_file(file_id, agent_id=agent_id)
            if not file_info:
                continue
            chunk_hits.append(self._format_chunk_hit(ch, relevance, file_info))

        self._enrich_hits_file_paths(chunk_hits)
        self._enrich_hits_preview_contract(chunk_hits)
        legacy_n = max(1, min(n, 50))
        results = chunk_hits[:legacy_n]
        print(
            f"📁✅ [FILE][semantic] chunk_hits={len(chunk_hits)} results={len(results)}"
        )
        return {"chunk_hits": chunk_hits, "results": results}
    
    def get_statistics(self) -> Dict:
        """Get document statistics"""
        folders = self.db_service.get_all_folders()
        files = self.db_service.get_all_files()
        chunks = self.vector_db.count() if self.vector_db else 0
        
        return {
            'total_folders': len(folders),
            'total_files': len(files),
            'total_chunks': chunks,
            'avg_chunks_per_file': chunks / len(files) if files else 0
        }
    
    def delete_document(self, file_id: str, *, preserve_source_path: bool = False) -> bool:
        """Delete a document and its associated files from training storage."""
        print(f"[FILE] delete_document start: {file_id}")
        file_info = self.db_service.get_file(file_id)
        if file_info:
            preserve = self._preserve_disk_paths(file_info, preserve_source_path)
            # Delete preview images (generated assets; never the origin file_path)
            try:
                if len(file_info) > 11 and file_info[11]:
                    self._remove_disk_file_unless_preserved(file_info[11], preserve)
                if len(file_info) > 12 and file_info[12]:
                    self._remove_disk_file_unless_preserved(file_info[12], preserve)
            except Exception as e:
                print(f"[FILE] Error deleting preview images for {file_id}: {e}")

            file_type = (file_info[3] or "").lower() if len(file_info) > 3 else ""
            if file_type in OFFICE_EXTENSIONS:
                cached_pdf = preview_pdf_path(file_id, DEFAULT_PREVIEW_PDF_CACHE_DIR)
                try:
                    if os.path.isfile(cached_pdf):
                        os.remove(cached_pdf)
                except OSError as e:
                    print(f"[FILE] Error deleting preview PDF for {file_id}: {e}")
            
            # Delete extracted images
            try:
                extracted_imgs = self.db_service.get_extracted_images(file_id)
                for img_info in extracted_imgs:
                    self._remove_disk_file_unless_preserved(img_info[1], preserve)
            except Exception as e:
                print(f"[FILE] Error deleting extracted images for {file_id}: {e}")
            
            # Delete from vector DB
            if self.vector_db:
                try:
                    self.vector_db.delete_document(file_id)
                except Exception as e:
                    print(f"[FILE] Error deleting vector embeddings for {file_id}: {e}")
            
            # Delete from sparse index
            if self.bm25_index:
                try:
                    self.bm25_index.delete_file_chunks(file_id)
                except Exception as e:
                    print(f"[FILE] Error deleting sparse index chunks for {file_id}: {e}")
            
            # Delete from database
            print(f"[FILE] Deleting SQL metadata for {file_id}")
            deleted = self.db_service.delete_file(file_id)
            if deleted:
                cleanup_paths = []
                if len(file_info) > 4 and file_info[4]:
                    disk_fp = os.path.normpath(str(file_info[4]))
                    cleanup_paths.append(disk_fp)
                    cleanup_paths.append(os.path.dirname(disk_fp))
                if len(file_info) > 5 and file_info[5]:
                    logical = str(file_info[5]).strip().strip("/\\")
                    if logical:
                        cleanup_paths.append(
                            os.path.normpath(os.path.join(self.upload_folder, logical))
                        )
                if cleanup_paths:
                    stats = clear_upload_scheduling_for_paths(
                        agent_id=self._ingestion_agent_id,
                        paths=cleanup_paths,
                        scan_metadata_db_file=getattr(self.db_service, "db_file", None)
                        or FILE_SYS_METADATA_DB,
                    )
                    print(
                        f"[FILE] Cleared upload scheduling after file delete: {stats}"
                    )
            return deleted
        
        print(f"[FILE] delete_document: file {file_id} not found in DB")
        return False
    
    def clear_all(self) -> bool:
        """Clear all documents and files"""
        if self.vector_db:
            self.vector_db.clear_all()
        
        if self.bm25_index:
            self.bm25_index.clear_all()
        
        result = self.db_service.clear_all()
        
        for folder in [self.upload_folder, self.image_folder, self.extracted_folder]:
            if os.path.exists(folder):
                for file in os.listdir(folder):
                    try:
                        os.remove(os.path.join(folder, file))
                    except:
                        pass
        
        return result
    
    def get_all_folders(self) -> List[tuple]:
        """Get all folders"""
        return self.db_service.get_all_folders()
    
    def delete_folder(self, folder_path: str) -> bool:
        """
        Delete a folder and all its contents (files, subfolders).
        Also cleans up vector embeddings and disk files.
        """
        folder_path = folder_path.strip()
        print(f"[FILE] delete_folder start: {folder_path}")
        
        # 1. Resolve canonical folder path (handle folder name vs full path)
        resolved_path = folder_path
        import os
        # Extract folder name in case a full local path was provided
        folder_name_only = os.path.basename(folder_path.rstrip('/').rstrip('\\'))
        
        all_folders = self.db_service.get_all_folders()
        found = False
        for f in all_folders:
            # f is (folder_name, folder_path, ...)
            # Check exact path match first
            if f[1] == folder_path:
                found = True
                break
            # Then check name match (either the input was the name, or we extracted it)
            if f[0] == folder_path or f[0] == folder_name_only:
                resolved_path = f[1]
                found = True
                print(f"[FILE] Resolved folder input '{folder_path}' to DB path '{resolved_path}'")
                break
        
        if found:
            folder_path = resolved_path

        # 2. Get all files in this folder (recursive) to delete from vector DB and disk
        files = self._get_all_files_in_folder_recursive(folder_path)
        print(f"[FILE] delete_folder: found {len(files)} files to clean up for path '{folder_path}'")
        
        # 3. Delete from vector DB and disk for each file
        for file_info in files:
            file_id = file_info[0]
            print(f"[FILE] Cleaning up file assets for {file_id}")
            
            # Delete preview images (indexes 11 and 12 in file_info tuple)
            try:
                if len(file_info) > 11 and file_info[11] and os.path.exists(file_info[11]):
                    os.remove(file_info[11])
                if len(file_info) > 12 and file_info[12] and os.path.exists(file_info[12]):
                    os.remove(file_info[12])
            except Exception as e:
                print(f"[FILE] Error deleting preview images for {file_id}: {e}")
            
            # Delete extracted images
            try:
                extracted_imgs = self.db_service.get_extracted_images(file_id)
                for img_info in extracted_imgs:
                    if os.path.exists(img_info[1]):
                        os.remove(img_info[1])
            except Exception as e:
                print(f"[FILE] Error deleting extracted images for {file_id}: {e}")
            
            # Delete from vector DB
            if self.vector_db:
                try:
                    self.vector_db.delete_document(file_id)
                    print(f"[FILE] Deleted vector embeddings for {file_id}")
                except Exception as e:
                    print(f"[FILE] Error deleting vector embeddings for {file_id}: {e}")
            
            # Delete from sparse index
            if self.bm25_index:
                try:
                    self.bm25_index.delete_file_chunks(file_id)
                    print(f"[FILE] Deleted sparse index chunks for {file_id}")
                except Exception as e:
                    print(f"[FILE] Error deleting sparse index chunks for {file_id}: {e}")
        
        # 4. Collect disk paths for schedule / scan-job cleanup (before SQL delete)
        cleanup_paths = set()
        logical = folder_path.strip().strip("/\\")
        if logical:
            cleanup_paths.add(
                os.path.normpath(os.path.join(self.upload_folder, logical))
            )
        cleanup_paths.add(os.path.normpath(self.upload_folder))
        for file_info in files:
            if len(file_info) > 4 and file_info[4]:
                disk_fp = os.path.normpath(str(file_info[4]))
                cleanup_paths.add(disk_fp)
                cleanup_paths.add(os.path.dirname(disk_fp))

        # 5. Finally delete from SQL database
        print(f"[FILE] Calling db_service.delete_folder for {folder_path}")
        deleted = self.db_service.delete_folder(folder_path)
        if deleted and cleanup_paths:
            stats = clear_upload_scheduling_for_paths(
                agent_id=self._ingestion_agent_id,
                paths=list(cleanup_paths),
                scan_metadata_db_file=getattr(self.db_service, "db_file", None)
                or FILE_SYS_METADATA_DB,
            )
            print(f"[FILE] Cleared upload scheduling after folder delete: {stats}")
        return deleted
    
    def get_files_in_folder(self, folder_path: str) -> List[Dict]:
        """Get all files in a folder as dictionaries"""
        files = self.db_service.get_files_in_folder(folder_path)
        
        results = []
        for file_info in files:
            results.append({
                'id': file_info[0],
                'filename': file_info[2],
                'file_type': file_info[3],
                'file_path': file_info[4],
                'folder_path': file_info[5],
                'relative_path': file_info[6],
                'upload_date': file_info[7],
                'size': file_info[8],
                'preview': file_info[9]
            })
        
        return results

    def get_files_in_folder_browse(self, folder_path: str) -> List[Dict]:
        """Browse listing: same dict shape as get_files_in_folder plus raw thumbnail_path (no full_content load)."""
        files = self.db_service.get_files_in_folder_browse(folder_path)
        results = []
        for file_info in files:
            results.append(
                {
                    "id": file_info[0],
                    "filename": file_info[2],
                    "file_type": file_info[3],
                    "file_path": file_info[4],
                    "folder_path": file_info[5],
                    "relative_path": file_info[6],
                    "upload_date": file_info[7],
                    "size": file_info[8],
                    "preview": file_info[9],
                    "thumbnail_path": file_info[11] if len(file_info) > 11 else "",
                }
            )
        return results

    def get_file_path(self, file_id: str) -> Optional[str]:
        """Get original file path for viewing/downloading (from files.file_path in metadata DB)."""
        path = self._file_path_from_db(file_id)
        return path if path and os.path.exists(path) else None
    
    def handle_query(self, query: str) -> Dict:
        """
        Handle document queries with @file, @analyze, @find prefixes.
        Migrated from original DocumentAssistant.
        """
        query = query.strip()
        trace = lf_client.trace(
            name="handle_query",
            input={"query": query},
            metadata={"prefix": query.split(" ", 1)[0].lower() if query else ""},
        )
        print(f"📁📝 [FILE][handle_query] start query='{query}'")
        try:
            if query.lower().startswith('@file'):
                search_query = query[5:].strip()
                print(f"📁📚 [FILE][@file] branch query='{search_query}'")
                search_out = self.search_semantic(search_query, n_results=10, trace=trace)
                chunk_hits = search_out.get("chunk_hits") or []
                results = search_out.get("results") or []
                self._enrich_hits_file_paths(chunk_hits)
                self._enrich_hits_file_paths(results)
                self._enrich_hits_preview_contract(chunk_hits)
                self._enrich_hits_preview_contract(results)
                folder_detected = self._detect_folder_from_query(search_query)

                folder_groups = {}
                for hit in chunk_hits:
                    folder = hit.get('folder_path') or 'Root'
                    if folder not in folder_groups:
                        folder_groups[folder] = []

                    file_key_findings = {}
                    folder_groups[folder].append({
                        'filename': hit['filename'],
                        'folder_path': hit.get('folder_path'),
                        'relative_path': hit.get('relative_path', ''),
                        'relevance': hit['relevance'],
                        'summary': (hit.get('chunk_text') or '')[:300],
                        'file_id': hit['file_id'],
                        'chunk_id': hit.get('chunk_id'),
                        'thumbnail': hit.get('thumbnail_path'),
                        'page_number': hit.get('page_number'),
                        'page_download_url': hit.get('page_download_url'),
                        'path': hit.get('path'),
                        'relevant_images': hit.get('relevant_images'),
                        'file_path': self._resolved_file_path_for_hit(hit),
                        'preview_url': hit.get('preview_url'),
                        'preview_content_type': hit.get('preview_content_type'),
                        'preview_mode': hit.get('preview_mode'),
                        'file_type': hit.get('file_type', 'Unknown'),
                        'key_findings': file_key_findings,
                        'agent_id': hit.get('agent_id'),
                    })

                overall_summary = ""
                if self.ai_analyzer:
                    overall_summary = self.ai_analyzer.summarize_search_results(
                        search_query,
                        chunk_hits,
                        folder_detected,
                        trace,
                    )

                payload = {
                    'type': 'document_search',
                    'query': search_query,
                    'folder_detected': folder_detected,
                    'chunk_hits': chunk_hits,
                    'folder_groups': folder_groups,
                    'results': results,
                    'overall_summary': overall_summary,
                    'matched_agent_ids': list(set(str(r.get('agent_id')) for r in chunk_hits if r.get('agent_id')))
                }
                print(
                    f"📁📚 [FILE][@file] done chunk_hits={len(chunk_hits)} results={len(results)} folder_detected='{folder_detected}'"
                )
                trace.update(output={"type": payload.get("type"), "results": len(results)})
                return payload

            elif query.lower().startswith('@analyze'):
                print("📁🧪 [FILE][@analyze] branch")
                raw = query[8:].strip()
                # Optional syntax: "@analyze <doc_ref> | <question>"
                doc_ref = raw
                analysis_question = ""
                if "||" in raw:
                    left, right = raw.split("||", 1)
                    doc_ref = left.strip()
                    analysis_question = right.strip()
                elif " | " in raw:
                    left, right = raw.split(" | ", 1)
                    doc_ref = left.strip()
                    analysis_question = right.strip()
                elif "|" in raw:
                    left, right = raw.split("|", 1)
                    doc_ref = left.strip()
                    analysis_question = right.strip()

                folder_path = self._detect_folder_from_query(doc_ref or raw)

                if folder_path:
                    files = self._get_all_files_in_folder_recursive(folder_path)
                else:
                    files = self.db_service.get_all_files()

                file_info = None

                # Try exact filename match
                for f in files:
                    if doc_ref.lower() == f[2].lower() or f[0] == doc_ref:
                        file_info = f
                        break

                # Try partial filename match
                if not file_info:
                    for f in files:
                        if doc_ref.lower() in f[2].lower():
                            file_info = f
                            break

                # Try semantic search
                if not file_info and self.vector_db:
                    # Filter vector search by current agent_id metadata
                    current_agent_id = self.agent_id
                    where_filter = None
                    if current_agent_id:
                        if isinstance(current_agent_id, list):
                            where_filter = {"agent_id": {"$in": [str(aid) for aid in current_agent_id]}}
                        else:
                            where_filter = {"agent_id": str(current_agent_id)}

                    search_results = self.vector_db.search(
                        doc_ref or raw,
                        n_results=5,
                        folder_path=folder_path,
                        where=where_filter
                    )

                    if search_results['ids'][0]:
                        best_meta = search_results['metadatas'][0][0]
                        best_file_id = best_meta.get('file_id')
                        best_agent_id = best_meta.get('agent_id')
                        file_info = self.db_service.get_file(best_file_id, agent_id=best_agent_id)

                if not file_info:
                    payload = {
                        'type': 'error',
                        'error': f'Document not found: {doc_ref or raw}',
                        'folder_detected': folder_path,
                        'files_searched': len(files)
                    }
                    trace.update(output={"type": payload.get("type"), "error": payload.get("error")})
                    return payload

                file_id = file_info[0]
                agent_id = file_info[13] if len(file_info) > 13 else None
                content = self.vector_db.get_document(file_id, agent_id=agent_id) if self.vector_db else ""
                images = self.db_service.get_extracted_images(file_info[0])

                analysis = None
                if self.ai_analyzer and content:
                    focused_content = content
                    if analysis_question and self.vector_db:
                        sub_queries = self.ai_analyzer.decompose_query(analysis_question, trace)
                        chunk_texts = []
                        for sq in sub_queries:
                            r = self.vector_db.search(
                                sq,
                                n_results=5,
                                where={"file_id": file_id},
                            )
                            if r and r.get("documents") and r["documents"] and r["documents"][0]:
                                chunk_texts.extend([t for t in r["documents"][0] if t])
                        # Dedup while preserving order
                        seen = set()
                        dedup = []
                        for t in chunk_texts:
                            key = t.strip()
                            if not key or key in seen:
                                continue
                            seen.add(key)
                            dedup.append(key)
                        if dedup:
                            focused_content = "\n\n---\n\n".join(dedup)[:12000]
                    analysis = self.ai_analyzer.analyze_document(focused_content, file_info[2], trace)

                payload = {
                    'type': 'document_analysis',
                    'file_id': file_id,
                    'filename': file_info[2],
                    'file_type': file_info[3],
                    'folder_path': file_info[5],
                    'relative_path': file_info[6],
                    'folder_detected': folder_path,
                    'analysis': analysis or 'AI analysis not available',
                    'all_images': [{'path': img[1], 'page': img[3], 'type': img[2]} for img in images],
                    'file_path': self._file_path_from_db(
                        file_info[0],
                        str(file_info[13]) if len(file_info) > 13 and file_info[13] else None,
                    ) or file_info[4],
                    'file_size': file_info[8],
                    'upload_date': file_info[7],
                    'agent_id': file_info[13] if len(file_info) > 13 else None,
                }
                if file_id and payload.get("agent_id") is not None:
                    apply_preview_contract_to_mapping(
                        payload,
                        file_id=str(file_id),
                        agent_id=str(payload["agent_id"]),
                        filename=str(payload.get("filename") or ""),
                        file_type=str(payload.get("file_type") or ""),
                        relative_url=True,
                    )
                print(
                    f"📁🧪 [FILE][@analyze] done filename='{payload.get('filename')}' folder_detected='{folder_path}'"
                )
                trace.update(output={"type": payload.get("type"), "filename": payload.get("filename")})
                return payload

            elif query.lower().startswith('@find'):
                print("📁🔍 [FILE][@find] branch")
                search_query = query[5:].strip()

                with timer("doc.find.detect_folder"):
                    folder_path = self._detect_folder_from_query(search_query)

                if folder_path:
                    with timer("doc.find.load_files_recursive"):
                        all_files = self._get_all_files_in_folder_recursive(folder_path)
                else:
                    with timer("doc.find.load_files_all"):
                        all_files = self.db_service.get_all_files()

                if not all_files:
                    payload = {
                        'type': 'document_find',
                        'query': search_query,
                        'folder_detected': folder_path,
                        'results': [],
                        'message': 'No files found',
                        'total_files_searched': 0
                    }
                    print("📁🔍 [FILE][@find] no files found in scope")
                    trace.update(output={"type": payload.get("type"), "results": 0})
                    return payload

                if not self.vector_db:
                    payload = {
                        'type': 'document_find',
                        'query': search_query,
                        'folder_detected': folder_path,
                        'results': [],
                        'message': 'Vector store not available',
                        'total_files_searched': len(all_files)
                    }
                    print("📁🔍 [FILE][@find] vector store unavailable")
                    trace.update(output={"type": payload.get("type"), "results": 0})
                    return payload

                # Use a composite key (file_id, agent_id) to avoid collisions in multi-agent search
                file_map = {(fi[0], str(fi[13]) if len(fi) > 13 else None): fi for fi in all_files if fi}

                # Hybrid Search: Semantic (Dense) + BM25 (Sparse)
                dense_ranked = []
                sparse_ranked = []
                chunk_map = {}

                with timer("doc.find.hybrid_search"):
                    # 1. Semantic Search
                    current_agent_id = self.agent_id
                    where_filter = None
                    if current_agent_id:
                        if isinstance(current_agent_id, list):
                            where_filter = {"agent_id": {"$in": [str(aid) for aid in current_agent_id]}}
                        else:
                            where_filter = {"agent_id": str(current_agent_id)}

                    semantic_results = self.vector_db.search(
                        search_query,
                        n_results=min(max(24, len(file_map) * 3), 120),
                        folder_path=folder_path,
                        where=where_filter,
                    )
                    
                    if semantic_results and semantic_results.get('ids') and semantic_results['ids'][0]:
                        for cid, doc, meta, dist in zip(
                            semantic_results['ids'][0],
                            semantic_results['documents'][0],
                            semantic_results['metadatas'][0],
                            semantic_results['distances'][0]
                        ):
                            dense_ranked.append(cid)
                            chunk_map[cid] = {'text': doc, 'metadata': meta, 'distance': dist}

                    # 2. BM25 Search (agent/folder filtering pushed into the search so
                    # the agent's true top-K survives at scale instead of being filtered
                    # out of a global top-K).
                    if self.bm25_index:
                        if isinstance(self.agent_id, list):
                            bm25_agent_ids = [str(a) for a in self.agent_id if a]
                        elif self.agent_id:
                            bm25_agent_ids = [str(self.agent_id)]
                        else:
                            bm25_agent_ids = None
                        bm25_hits = self.bm25_index.search(
                            search_query,
                            n_results=40,
                            agent_ids=bm25_agent_ids,
                            folder_path=folder_path,
                        )
                        for h in bm25_hits:
                            sparse_ranked.append(h.chunk_id)
                            if h.chunk_id not in chunk_map:
                                chunk_map[h.chunk_id] = {'text': h.document, 'metadata': h.metadata, 'bm25_score': h.score}

                # 3. Fusion (RRF)
                from app.services.document_nodes import reciprocal_rank_fusion
                fused_ids = reciprocal_rank_fusion([dense_ranked, sparse_ranked], k=60)

                find_chunk_objs: List[DocumentChunk] = []
                for cid in fused_ids[:TOP_CHUNK_HITS]:
                    cdata = chunk_map.get(cid)
                    if not cdata:
                        continue
                    meta = cdata.get("metadata") or {}
                    find_chunk_objs.append(
                        DocumentChunk(
                            chunk_id=str(cid),
                            text=cdata.get("text") or "",
                            metadata=meta,
                            distance=cdata.get("distance"),
                            bm25_score=cdata.get("bm25_score"),
                        )
                    )
                find_rels = chunk_relevance_scores_0_100(find_chunk_objs, query=search_query)
                chunk_hits: List[Dict] = []
                for ch, rel in zip(find_chunk_objs, find_rels):
                    meta = ch.metadata or {}
                    fid = meta.get("file_id")
                    a_id = meta.get("agent_id")
                    fi = file_map.get((fid, str(a_id) if a_id else None)) if fid else None
                    if not fi:
                        continue
                    chunk_hits.append(self._format_chunk_hit(ch, rel, fi))
                
                # Group chunks by file (using composite key)
                chunks_by_file: Dict[Tuple[str, str], List[Dict]] = {}
                for cid in fused_ids:
                    cdata = chunk_map.get(cid)
                    if not cdata: continue
                    meta = cdata.get('metadata') or {}
                    f_id = meta.get('file_id')
                    a_id = str(meta.get('agent_id')) if meta.get('agent_id') else None
                    
                    composite_key = (f_id, a_id)
                    if not f_id or composite_key not in file_map:
                        continue
                    
                    # Calculate a combined relevance score
                    if 'distance' in cdata:
                        chunk_relevance = max(0, 100 - (cdata['distance'] * 100))
                    elif 'bm25_score' in cdata:
                        chunk_relevance = min(95, 20 + (cdata['bm25_score'] * 5))
                    else:
                        chunk_relevance = 50.0
                        
                    chunks_by_file.setdefault(composite_key, []).append({
                        'text': cdata['text'],
                        'relevance': chunk_relevance,
                        'chunk_id': cid,
                        'metadata': meta,
                    })

                candidate_keys = list(chunks_by_file.keys())
                
                # Enhanced Filename matching (word-based)
                search_words = set(re.findall(r'\w+', search_query.lower()))
                for ck, fi in file_map.items():
                    if ck not in candidate_keys:
                        fname = str(fi[2]).lower()
                        fname_words = set(re.findall(r'\w+', fname))
                        # If any word from query is in filename, or substring match
                        if search_words.intersection(fname_words) or search_query.lower() in fname:
                            candidate_keys.append(ck)
                            chunks_by_file[ck] = [{'text': f"Filename match: {fi[2]}", 'relevance': 90.0}]

                if not candidate_keys:
                    payload = {
                        'type': 'document_find',
                        'query': search_query,
                        'folder_detected': folder_path,
                        'results': [],
                        'message': 'No relevant matches found',
                        'total_files_searched': len(all_files)
                    }
                    print("📁🔍 [FILE][@find] no semantic or filename candidate files")
                    trace.update(output={"type": payload.get("type"), "results": 0})
                    return payload

                def _score_candidate_file(composite_key: Tuple[str, str]):
                    file_info_tuple = file_map.get(composite_key)
                    file_chunks = chunks_by_file.get(composite_key, [])[:3]
                    if not file_info_tuple or not file_chunks:
                        return None

                    file_relevance = sum(c['relevance'] for c in file_chunks) / len(file_chunks)
                    with timer("doc.find.get_document"):
                        # Chroma search for full document using file_id AND agent_id for strict isolation
                        content = self.vector_db.get_document(composite_key[0], agent_id=composite_key[1])
                    if not content:
                        return None

                    search_lower = search_query.lower()
                    content_lower = content.lower()
                    exact_matches = content_lower.count(search_lower)
                    if exact_matches > 0:
                        file_relevance = min(100, file_relevance + (exact_matches * 5))

                    return {
                        'file_info': file_info_tuple,
                        'relevance': file_relevance,
                        'exact_matches': exact_matches,
                        'best_chunks': file_chunks,
                        'content': content,
                    }

                file_results = []
                max_workers = min(8, max(1, len(candidate_keys)))
                with timer(f"doc.find.parallel_score(files={len(candidate_keys)},workers={max_workers})"):
                    with ThreadPoolExecutor(max_workers=max_workers) as pool:
                        futures = [pool.submit(_score_candidate_file, ck) for ck in candidate_keys]
                        for fut in as_completed(futures):
                            try:
                                r = fut.result()
                                if r:
                                    file_results.append(r)
                            except Exception:
                                continue

                file_results.sort(key=lambda x: x['relevance'], reverse=True)

                if not file_results or file_results[0]['relevance'] < 5:
                    payload = {
                        'type': 'document_find',
                        'query': search_query,
                        'folder_detected': folder_path,
                        'results': [],
                        'message': 'No relevant matches found',
                        'total_files_searched': len(all_files)
                    }
                    print("📁🔍 [FILE][@find] top relevance below threshold")
                    trace.update(output={"type": payload.get("type"), "results": 0})
                    return payload

                best = file_results[0]
                file_info = best['file_info']
                content = best['content']

                lines = content.split('\n')
                matching_sections = []
                search_lower = search_query.lower()

                for i, line in enumerate(lines):
                    if search_lower in line.lower():
                        start = max(0, i - 8)
                        end = min(len(lines), i + 9)
                        context_lines = lines[start:end]

                        context_text = []
                        for j, ctx_line in enumerate(context_lines):
                            if start + j == i:
                                context_text.append(f">>> {ctx_line} <<<")
                            else:
                                context_text.append(ctx_line)

                        context = '\n'.join(context_text)

                        if len(matching_sections) < 10:
                            matching_sections.append(context)

                # Page citations from chunk metadata (PDF preview via page_number).
                citation_pages: set = set()
                image_source_chunks = [
                    c
                    for c in best["best_chunks"]
                    if search_lower in (c.get("text") or "").lower()
                ] or list(best["best_chunks"])
                for c in image_source_chunks:
                    meta = c.get("metadata") or {}
                    pn = page_number_from_metadata(meta)
                    if pn is not None:
                        citation_pages.add(pn)

                for h in chunk_hits:
                    if h.get("file_id") != file_info[0]:
                        continue
                    pn = h.get("page_number")
                    if pn is not None:
                        citation_pages.add(int(pn))

                matching_pages = sorted(citation_pages)
                find_agent_id = (
                    str(file_info[13]) if len(file_info) > 13 and file_info[13] else None
                ) or str(self._ingestion_agent_id or "")
                matching_page_urls = build_matching_page_urls(
                    file_id=str(file_info[0]),
                    agent_id=find_agent_id,
                    pages=matching_pages,
                    relative_url=True,
                )
                page_images = build_page_images_from_api_paths(
                    str(file_info[0]),
                    find_agent_id,
                    matching_pages,
                )

                ai_summary = None
                if self.ai_analyzer:
                    chunks_text = '\n\n'.join([c['text'] for c in best['best_chunks']])
                    ai_summary = self.ai_analyzer.find_in_document(
                        search_query,
                        chunks_text,
                        file_info[2],
                        trace,
                    )

                best_result = {
                    'filename': file_info[2],
                    'file_id': file_info[0],
                    'folder_path': file_info[5],
                    'relative_path': file_info[6],
                    'match_count': best['exact_matches'] or len(matching_sections),
                    'relevance_score': round(best['relevance'], 1),
                    'matches': matching_sections,
                    'matching_pages': sorted(matching_pages),
                    'matching_page_urls': matching_page_urls,
                    'page_images': page_images,
                    'file_path': self._file_path_from_db(
                        file_info[0],
                        str(file_info[13]) if len(file_info) > 13 and file_info[13] else None,
                    ) or file_info[4],
                    'search_term': search_query,
                    'ai_summary': ai_summary,
                    'top_chunks': [
                        {'text': c.get('text', ''), 'relevance': c.get('relevance', 0)}
                        for c in best['best_chunks'][:3]
                    ],
                    'agent_id': file_info[13] if len(file_info) > 13 else None,
                }
                find_agent_id = best_result.get("agent_id")
                if best_result.get("file_id") and find_agent_id is not None:
                    apply_preview_contract_to_mapping(
                        best_result,
                        file_id=str(best_result["file_id"]),
                        agent_id=str(find_agent_id),
                        filename=str(best_result.get("filename") or file_info[2]),
                        file_type=str(file_info[3] if len(file_info) > 3 else ""),
                        relative_url=True,
                    )
                if not best_result.get("file_type") and len(file_info) > 3:
                    best_result["file_type"] = file_info[3]

                # Collect all agents that have a file with the same top relevance (handles identical files across agents)
                best_relevance = best['relevance']
                matched_agents = set()
                for r in file_results:
                    if r['relevance'] >= best_relevance - 0.1:
                        aid = r['file_info'][13] if len(r['file_info']) > 13 else None
                        if aid:
                            matched_agents.add(str(aid))
                    else:
                        break

                payload = {
                    'type': 'document_find',
                    'query': search_query,
                    'folder_detected': folder_path,
                    'results': [best_result],
                    'total_files_searched': len(all_files),
                    'matched_agent_ids': sorted(list(matched_agents))
                }
                print(
                    f"📁🔍 [FILE][@find] done searched={len(all_files)} matches={len(payload.get('results') or [])}"
                )
                trace.update(output={"type": payload.get("type"), "results": len(payload.get("results") or [])})
                return payload

            else:
                # General semantic search
                print("📁🌐 [FILE][general] semantic search branch")
                search_out = self.search_semantic(query, n_results=5, trace=trace)
                chunk_hits = search_out.get("chunk_hits") or []
                results = search_out.get("results") or []
                self._enrich_hits_file_paths(chunk_hits)
                self._enrich_hits_file_paths(results)
                self._enrich_hits_preview_contract(chunk_hits)
                self._enrich_hits_preview_contract(results)
                folder_detected = self._detect_folder_from_query(query)

                folder_groups = {}
                for hit in chunk_hits:
                    folder = hit.get('folder_path') or 'Root'
                    if folder not in folder_groups:
                        folder_groups[folder] = []

                    folder_groups[folder].append({
                        'filename': hit['filename'],
                        'folder_path': hit.get('folder_path'),
                        'relative_path': hit.get('relative_path', ''),
                        'relevance': hit['relevance'],
                        'summary': (hit.get('chunk_text') or '')[:300],
                        'file_id': hit['file_id'],
                        'chunk_id': hit.get('chunk_id'),
                        'thumbnail': hit.get('thumbnail_path'),
                        'page_number': hit.get('page_number'),
                        'page_download_url': hit.get('page_download_url'),
                        'path': hit.get('path'),
                        'relevant_images': hit.get('relevant_images'),
                        'file_path': self._resolved_file_path_for_hit(hit),
                        'preview_url': hit.get('preview_url'),
                        'preview_content_type': hit.get('preview_content_type'),
                        'preview_mode': hit.get('preview_mode'),
                        'file_type': hit.get('file_type', 'Unknown'),
                        'agent_id': hit.get('agent_id'),
                    })

                overall_summary = ""
                if self.ai_analyzer:
                    overall_summary = self.ai_analyzer.summarize_search_results(
                        query, chunk_hits, folder_detected, trace
                    )

                payload = {
                    'type': 'document_search',
                    'query': query,
                    'folder_detected': folder_detected,
                    'chunk_hits': chunk_hits,
                    'folder_groups': folder_groups,
                    'results': results,
                    'overall_summary': overall_summary,
                    'matched_agent_ids': list(set(str(r.get('agent_id')) for r in chunk_hits if r.get('agent_id')))
                }
                trace.update(output={"type": payload.get("type"), "results": len(results)})
                return payload
        except Exception as e:
            trace.update(level="ERROR", status_message=str(e))
            raise
        finally:
            lf_client.flush()
