"""
Activity-based news subscription scheduler for per-user encrypted databases.
Tracks user activity and temporarily stores credentials for automatic updates.
"""

import random
import threading
from dataclasses import dataclass
from datetime import datetime, timedelta, UTC
from functools import wraps
from typing import Any, Callable, Dict, List

from cachetools import TTLCache
from loguru import logger
from ..settings.logger import log_settings
from ..settings.manager import SnapshotSettingsContext

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.base import JobLookupError
from sqlalchemy import func
from ..constants import ResearchStatus
from ..database.credential_store_base import CredentialStoreBase
from ..database.session_context import safe_rollback
from ..database.thread_local_session import thread_cleanup
from ..security.log_sanitizer import redact_secrets

# RAG indexing imports. The reconciler builds RAG services via the lazily
# imported ``rag_service_factory`` (see _reconcile_unindexed_documents), so the
# concrete ``LibraryRAGService`` is no longer imported at module scope.
from ..database.library_init import get_default_library_id
from ..database.models.library import Document, DocumentCollection
from ..constants import DEFAULT_SEARCH_TOOL


SCHEDULER_AVAILABLE = True  # Always available since it's a required dependency

# Per-tick cap for the opt-in library-collection index sweep. Bounds the work
# done in a single scheduler thread tick so a large backlog of unindexed
# documents self-heals gradually over successive ticks instead of blocking the
# worker thread (the sweep is self-rate-limited by APScheduler max_instances=1
# plus this batch cap).
_LIBRARY_SWEEP_BATCH = 50


class SchedulerCredentialStore(CredentialStoreBase):
    """Credential store for the news scheduler.

    Stores user passwords with TTL expiration so that background scheduler
    jobs can access encrypted per-user databases.
    """

    def __init__(self, ttl_hours: int = 48):
        super().__init__(ttl_hours * 3600)

    def store(self, username: str, password: str) -> None:
        """Store password for a user."""
        self._store_credentials(
            username, {"username": username, "password": password}
        )

    def retrieve(self, username: str) -> str | None:
        """Retrieve password for a user. Returns None if expired/missing."""
        result = self._retrieve_credentials(username, remove=False)
        return result[1] if result else None

    def clear(self, username: str) -> None:
        """Clear stored password for a user."""
        self.clear_entry(username)


@dataclass(frozen=True)
class DocumentSchedulerSettings:
    """
    Immutable settings snapshot for document scheduler.

    Thread-safe: This is a frozen dataclass that can be safely passed
    to and used from background threads.
    """

    enabled: bool = True
    interval_seconds: int = 1800
    download_pdfs: bool = False
    extract_text: bool = True
    generate_rag: bool = False
    sweep_library_collections: bool = False
    last_run: str = ""

    @classmethod
    def defaults(cls) -> "DocumentSchedulerSettings":
        """Return default settings."""
        return cls()


class BackgroundJobScheduler:
    """
    Singleton scheduler that manages news subscriptions for active users.

    This scheduler:
    - Monitors user activity through database access
    - Temporarily stores user credentials in memory
    - Automatically schedules subscription checks
    - Cleans up inactive users after configurable period
    """

    _instance = None
    _lock = threading.Lock()

    def __new__(cls):
        """Ensure singleton instance."""
        if cls._instance is None:
            with cls._lock:
                if cls._instance is None:
                    cls._instance = super().__new__(cls)
        return cls._instance

    def __init__(self):
        """Initialize the scheduler (only runs once due to singleton)."""
        # Skip if already initialized
        if hasattr(self, "_initialized"):
            return

        # User session tracking
        self.user_sessions = {}  # user_id -> {last_activity, scheduled_jobs}
        self.lock = threading.Lock()

        # Credential store with TTL-based expiration
        self._credential_store = SchedulerCredentialStore(ttl_hours=48)

        # Scheduler instance
        self.scheduler = BackgroundScheduler()

        # Configuration (will be loaded from settings)
        self.config = self._load_default_config()

        # State
        self.is_running = False
        self._app = None  # Flask app reference for background job contexts

        # Settings cache: username -> DocumentSchedulerSettings
        # TTL of 300 seconds (5 minutes) reduces database queries
        self._settings_cache: TTLCache = TTLCache(maxsize=100, ttl=300)
        self._settings_cache_lock = threading.Lock()

        self._initialized = True
        logger.info("News scheduler initialized")

    def _load_default_config(self) -> Dict[str, Any]:
        """Load default configuration (will be overridden by settings manager)."""
        return {
            "enabled": True,
            "retention_hours": 48,
            "cleanup_interval_hours": 1,
            "max_jitter_seconds": 300,
            "max_concurrent_jobs": 10,
            "subscription_batch_size": 5,
            "activity_check_interval_minutes": 5,
        }

    def initialize_with_settings(self, settings_manager):
        """Initialize configuration from settings manager."""
        try:
            # Load all scheduler settings
            self.settings_manager = settings_manager
            self.config = {
                "enabled": self._get_setting("news.scheduler.enabled", True),
                "retention_hours": self._get_setting(
                    "news.scheduler.retention_hours", 48
                ),
                "cleanup_interval_hours": self._get_setting(
                    "news.scheduler.cleanup_interval_hours", 1
                ),
                "max_jitter_seconds": self._get_setting(
                    "news.scheduler.max_jitter_seconds", 300
                ),
                "max_concurrent_jobs": self._get_setting(
                    "news.scheduler.max_concurrent_jobs", 10
                ),
                "subscription_batch_size": self._get_setting(
                    "news.scheduler.batch_size", 5
                ),
                "activity_check_interval_minutes": self._get_setting(
                    "news.scheduler.activity_check_interval", 5
                ),
            }
            log_settings(self.config, "Scheduler configuration loaded")
        except Exception:
            logger.exception("Error loading scheduler settings")
            # Keep default config

    def _get_setting(self, key: str, default: Any) -> Any:
        """Get setting with fallback to default."""
        if hasattr(self, "settings_manager") and self.settings_manager:
            return self.settings_manager.get_setting(key, default=default)
        return default

    def set_app(self, app) -> None:
        """Store a reference to the Flask app for creating app contexts in background jobs."""
        self._app = app

    def _wrap_job(self, func: Callable) -> Callable:
        """Wrap a scheduler job function so it runs inside a Flask app context.

        APScheduler runs jobs in a thread pool without Flask context.
        This wrapper pushes an app context before the job runs and pops it after.
        """

        @wraps(func)
        def wrapper(*args, **kwargs):
            if self._app is not None:
                with self._app.app_context():
                    return func(*args, **kwargs)
            else:
                logger.warning(
                    f"No Flask app set on scheduler; running {func.__name__} without app context"
                )
                return func(*args, **kwargs)

        return wrapper

    def _get_document_scheduler_settings(
        self, username: str, force_refresh: bool = False
    ) -> DocumentSchedulerSettings:
        """
        Get document scheduler settings for a user with TTL caching.

        This is the single source of truth for document scheduler settings.
        Settings are cached for 5 minutes by default to reduce database queries.

        Args:
            username: User to get settings for
            force_refresh: If True, bypass cache and fetch fresh settings

        Returns:
            DocumentSchedulerSettings dataclass (frozen/immutable for thread-safety)
        """
        # Fast path: check cache without modifying it
        if not force_refresh:
            with self._settings_cache_lock:
                cached = self._settings_cache.get(username)
                if cached is not None:
                    logger.debug(f"[SETTINGS_CACHE] Cache hit for {username}")
                    cached_settings: DocumentSchedulerSettings = cached
                    return cached_settings

        # Cache miss - need to fetch from database
        logger.debug(
            f"[SETTINGS_CACHE] Cache miss for {username}, fetching from DB"
        )

        # Get password from session
        session_info = self.user_sessions.get(username)
        if not session_info:
            logger.warning(
                f"[SETTINGS_CACHE] No session info for {username}, using defaults"
            )
            return DocumentSchedulerSettings.defaults()

        password = self._credential_store.retrieve(username)
        if not password:
            logger.warning(
                f"[SETTINGS_CACHE] Credentials expired for {username}, using defaults"
            )
            return DocumentSchedulerSettings.defaults()

        # Fetch settings from database (outside lock to avoid blocking)
        try:
            from ..database.session_context import get_user_db_session
            from ..settings.manager import SettingsManager

            with get_user_db_session(username, password) as db:
                sm = SettingsManager(db)

                settings = DocumentSchedulerSettings(
                    enabled=sm.get_setting("document_scheduler.enabled", True),
                    interval_seconds=sm.get_setting(
                        "document_scheduler.interval_seconds", 1800
                    ),
                    download_pdfs=sm.get_setting(
                        "document_scheduler.download_pdfs", False
                    ),
                    extract_text=sm.get_setting(
                        "document_scheduler.extract_text", True
                    ),
                    generate_rag=sm.get_setting(
                        "document_scheduler.generate_rag", False
                    ),
                    sweep_library_collections=sm.get_setting(
                        "document_scheduler.sweep_library_collections", False
                    ),
                    last_run=sm.get_setting("document_scheduler.last_run", ""),
                )

            # Store in cache
            with self._settings_cache_lock:
                self._settings_cache[username] = settings
                logger.debug(f"[SETTINGS_CACHE] Cached settings for {username}")

            return settings

        except Exception as e:
            # ``password`` (retrieved above) is in the frame locals of
            # this handler and of ``get_user_db_session``; a traceback
            # rendered with diagnose=True would leak it. Drop the
            # traceback chain and redact str(e).
            safe_msg = redact_secrets(str(e), password)
            logger.warning(
                f"[SETTINGS_CACHE] Error fetching settings for {username}: {safe_msg}"
            )
            return DocumentSchedulerSettings.defaults()

    def invalidate_user_settings_cache(self, username: str) -> bool:
        """
        Invalidate cached settings for a specific user.

        Call this when user settings change or user logs out.

        Args:
            username: User whose cache to invalidate

        Returns:
            True if cache entry was removed, False if not found
        """
        with self._settings_cache_lock:
            if username in self._settings_cache:
                del self._settings_cache[username]
                logger.debug(
                    f"[SETTINGS_CACHE] Invalidated cache for {username}"
                )
                return True
            return False

    def invalidate_all_settings_cache(self) -> int:
        """
        Invalidate all cached settings.

        Call this when doing bulk settings updates or during config reload.

        Returns:
            Number of cache entries cleared
        """
        with self._settings_cache_lock:
            count = len(self._settings_cache)
            self._settings_cache.clear()
            logger.info(
                f"[SETTINGS_CACHE] Cleared all settings cache ({count} entries)"
            )
            return count

    def start(self):
        """Start the scheduler."""
        if not self.config.get("enabled", True):
            logger.info("News scheduler is disabled in settings")
            return

        if self.is_running:
            logger.warning("Scheduler is already running")
            return

        if self._app is None:
            raise RuntimeError(
                "BackgroundJobScheduler.set_app() must be called before start()"
            )

        # Schedule cleanup job
        self.scheduler.add_job(
            self._wrap_job(self._run_cleanup_with_tracking),
            "interval",
            hours=self.config["cleanup_interval_hours"],
            id="cleanup_inactive_users",
            name="Cleanup Inactive User Sessions",
            jitter=60,  # Add some jitter to cleanup
        )

        # Schedule configuration reload
        self.scheduler.add_job(
            self._wrap_job(self._reload_config),
            "interval",
            minutes=30,
            id="reload_config",
            name="Reload Configuration",
        )

        # Start the scheduler
        self.scheduler.start()
        self.is_running = True

        # Schedule initial cleanup after a delay
        self.scheduler.add_job(
            self._wrap_job(self._run_cleanup_with_tracking),
            "date",
            run_date=datetime.now(UTC) + timedelta(seconds=30),
            id="initial_cleanup",
        )

        logger.info("News scheduler started")

    def stop(self):
        """Stop the scheduler."""
        if self.is_running:
            self.scheduler.shutdown(wait=True)
            self.is_running = False

            # Clear all user sessions and credentials
            with self.lock:
                for username in self.user_sessions:
                    self._credential_store.clear(username)
                self.user_sessions.clear()

            logger.info("News scheduler stopped")

    def update_user_info(self, username: str, password: str):
        """
        Update user info in scheduler. Called on every database interaction.

        Args:
            username: User's username
            password: User's password
        """
        logger.info(
            f"[SCHEDULER] update_user_info called for {username}, is_running={self.is_running}, active_users={len(self.user_sessions)}"
        )
        logger.debug(
            f"[SCHEDULER] Current active users: {list(self.user_sessions.keys())}"
        )

        if not self.is_running:
            logger.warning(
                f"[SCHEDULER] Scheduler not running, cannot update user {username}"
            )
            return

        with self.lock:
            # Store password in credential store (inside lock to prevent
            # race where concurrent calls leave mismatched credentials)
            self._credential_store.store(username, password)

            now = datetime.now(UTC)

            if username not in self.user_sessions:
                # New user - create session info
                logger.info(f"[SCHEDULER] New user in scheduler: {username}")
                self.user_sessions[username] = {
                    "last_activity": now,
                    "scheduled_jobs": set(),
                }
                logger.debug(
                    f"[SCHEDULER] Created session for {username}, scheduling subscriptions"
                )
                # Schedule their subscriptions
                self._schedule_user_subscriptions(username)
            else:
                # Existing user - update info
                logger.info(
                    f"[SCHEDULER] Updating existing user {username} activity, will reschedule"
                )
                old_activity = self.user_sessions[username]["last_activity"]
                activity_delta = now - old_activity
                logger.debug(
                    f"[SCHEDULER] User {username} last activity: {old_activity}, delta: {activity_delta}"
                )

                self.user_sessions[username]["last_activity"] = now
                logger.debug(
                    f"[SCHEDULER] Updated {username} session info, scheduling subscriptions"
                )
                # Reschedule their subscriptions in case they changed
                self._schedule_user_subscriptions(username)

    def unregister_user(self, username: str):
        """
        Unregister a user and clean up their scheduled jobs.
        Called when user logs out.
        """
        with self.lock:
            if username in self.user_sessions:
                logger.info(f"Unregistering user {username}")

                # Remove all scheduled jobs for this user
                session_info = self.user_sessions[username]
                for job_id in session_info["scheduled_jobs"].copy():
                    try:
                        self.scheduler.remove_job(job_id)
                    except JobLookupError:
                        pass

                # Remove user session and clear credentials atomically
                del self.user_sessions[username]
                self._credential_store.clear(username)

        # Invalidate settings cache for this user (outside lock)
        self.invalidate_user_settings_cache(username)
        logger.info(f"User {username} unregistered successfully")

    def reschedule_document_jobs(self, username: str) -> bool:
        """(Re)schedule the document-processing + reconciler jobs for an
        ACTIVE user against their current settings.

        Call this after a ``document_scheduler.*`` setting changes so the
        change takes effect on the next interval tick instead of only after
        the user logs out and back in. Without it, ``_schedule_reconciler``
        runs only via the login path (``update_user_info`` ->
        ``_schedule_user_subscriptions`` -> ``_schedule_document_processing``):
        the runtime gate inside ``_reconcile_unindexed_documents`` neutralises
        a stale job after a *disable*, but an *enable* (including toggling the
        legacy ``generate_rag`` arm, which on older builds took effect on the
        next tick) would otherwise never create the ``{username}_library_sweep``
        job until the next login.

        Relies on the cache having already been invalidated by the caller
        (``invalidate_settings_caches``) so ``_schedule_document_processing``
        re-reads fresh settings. No password argument is needed: the job is
        rebuilt from the credentials the scheduler already holds for the active
        session. Returns ``True`` if a reschedule was performed, ``False`` for
        a user the scheduler isn't tracking (their jobs are built from current
        settings on their next login) or when the scheduler isn't running.
        """
        if not self.is_running:
            return False
        with self.lock:
            if username not in self.user_sessions:
                logger.debug(
                    f"[DOC_SCHEDULER] reschedule_document_jobs: {username} "
                    "not an active scheduler session; skipping"
                )
                return False
            self.user_sessions[username]["last_activity"] = datetime.now(UTC)
            # Re-reads fresh settings (cache invalidated by the caller) and
            # re-adds the document-processing job + reconciler per current
            # settings; both use replace_existing=True so this is idempotent.
            self._schedule_document_processing(username)
        return True

    def _schedule_user_subscriptions(self, username: str):
        """Schedule all active subscriptions for a user."""
        logger.info(f"_schedule_user_subscriptions called for {username}")
        # Pre-declared so the leak-redaction in the except handler is safe
        # if the exception fires before ``password`` is assigned below.
        password = None
        try:
            session_info = self.user_sessions.get(username)
            if not session_info:
                logger.warning(f"No session info found for {username}")
                return

            password = self._credential_store.retrieve(username)
            if not password:
                logger.warning(
                    f"Credentials expired for {username}, skipping subscription scheduling"
                )
                return
            logger.debug(f"Got password for {username}: present")

            # Get user's subscriptions from their encrypted database
            from ..database.session_context import get_user_db_session
            from ..database.models.news import NewsSubscription

            with get_user_db_session(username, password) as db:
                subscriptions = (
                    db.query(NewsSubscription)
                    .filter(NewsSubscription.active_filter())
                    .all()
                )
                logger.debug(
                    f"Query executed, found {len(subscriptions)} results"
                )

                # Log details of each subscription
                for sub in subscriptions:
                    logger.debug(
                        f"Subscription {sub.id}: name='{sub.name}', status='{sub.status}', refresh_interval={sub.refresh_interval_minutes} minutes"
                    )

            logger.info(
                f"Found {len(subscriptions)} active subscriptions for {username}"
            )

            # Clear old jobs for this user
            for job_id in session_info["scheduled_jobs"].copy():
                try:
                    self.scheduler.remove_job(job_id)
                    session_info["scheduled_jobs"].remove(job_id)
                except JobLookupError:
                    pass

            # Schedule each subscription with jitter
            for sub in subscriptions:
                job_id = f"{username}_{sub.id}"

                # Calculate jitter
                # Security: random jitter to distribute subscription timing, not security-sensitive
                max_jitter = int(self.config.get("max_jitter_seconds", 300))
                jitter = random.randint(0, max_jitter)

                # Determine trigger based on frequency
                refresh_minutes = sub.refresh_interval_minutes

                if refresh_minutes <= 60:  # 60 minutes or less
                    # For hourly or more frequent, use interval trigger
                    trigger = "interval"
                    trigger_args = {
                        "minutes": refresh_minutes,
                        "jitter": jitter,
                        "start_date": datetime.now(UTC),  # Start immediately
                    }
                else:
                    # For less frequent, calculate next run time
                    now = datetime.now(UTC)
                    if sub.next_refresh:
                        # Ensure timezone-aware for comparison with now (UTC)
                        next_refresh_aware = sub.next_refresh
                        if next_refresh_aware.tzinfo is None:
                            logger.warning(
                                f"Subscription {sub.id} has naive (non-tz-aware) "
                                f"next_refresh datetime, assuming UTC"
                            )
                            next_refresh_aware = next_refresh_aware.replace(
                                tzinfo=UTC
                            )
                        if next_refresh_aware <= now:
                            # Subscription is overdue - run it immediately with small jitter
                            logger.info(
                                f"Subscription {sub.id} is overdue, scheduling immediate run"
                            )
                            next_run = now + timedelta(seconds=jitter)
                        else:
                            next_run = next_refresh_aware
                    else:
                        next_run = now + timedelta(
                            minutes=refresh_minutes, seconds=jitter
                        )

                    trigger = "date"
                    trigger_args = {"run_date": next_run}

                # Add the job
                self.scheduler.add_job(
                    func=self._wrap_job(self._check_subscription),
                    args=[username, sub.id],
                    trigger=trigger,
                    id=job_id,
                    name=f"Check {sub.name or sub.query_or_topic[:30]}",
                    replace_existing=True,
                    **trigger_args,
                )

                session_info["scheduled_jobs"].add(job_id)
                logger.info(f"Scheduled job {job_id} with {trigger} trigger")

        except Exception as e:
            # ``password`` was retrieved from the credential store
            # above (line ~483) and passed into ``get_user_db_session``.
            # An exception from the DB session (e.g. SQLCipher
            # ``OperationalError``) can carry frame locals that include
            # the plaintext SQLCipher master password — which is
            # unrecoverable (TRUST.md §5). Drop the traceback chain and
            # redact str(e).
            safe_msg = redact_secrets(str(e), password)
            logger.warning(
                f"Error scheduling subscriptions for {username}: {safe_msg}"
            )

        # Add document processing for this user
        self._schedule_document_processing(username)

    def _schedule_document_processing(self, username: str):
        """Schedule document processing for a user."""
        logger.info(
            f"[DOC_SCHEDULER] Scheduling document processing for {username}"
        )
        logger.debug(
            f"[DOC_SCHEDULER] Current user sessions: {list(self.user_sessions.keys())}"
        )

        try:
            session_info = self.user_sessions.get(username)
            if not session_info:
                logger.warning(
                    f"[DOC_SCHEDULER] No session info found for {username}"
                )
                logger.debug(
                    f"[DOC_SCHEDULER] Available sessions: {list(self.user_sessions.keys())}"
                )
                return

            logger.debug(
                f"[DOC_SCHEDULER] Retrieved session for {username}, scheduler running: {self.is_running}"
            )

            # Get user's document scheduler settings (cached)
            settings = self._get_document_scheduler_settings(username)

            if not settings.enabled:
                logger.info(
                    f"[DOC_SCHEDULER] Document scheduler disabled for user {username}"
                )
                return

            logger.info(
                f"[DOC_SCHEDULER] User {username} document settings: enabled={settings.enabled}, "
                f"interval={settings.interval_seconds}s, pdfs={settings.download_pdfs}, "
                f"text={settings.extract_text}, "
                f"index={settings.generate_rag or settings.sweep_library_collections}"
            )

            # Schedule document processing job
            job_id = f"{username}_document_processing"
            logger.debug(f"[DOC_SCHEDULER] Preparing to schedule job {job_id}")

            # Remove existing document job if any
            try:
                self.scheduler.remove_job(job_id)
                session_info["scheduled_jobs"].discard(job_id)
                logger.debug(f"[DOC_SCHEDULER] Removed existing job {job_id}")
            except JobLookupError:
                logger.debug(
                    f"[DOC_SCHEDULER] No existing job {job_id} to remove"
                )
                pass  # Job doesn't exist, that's fine

            # Add new document processing job
            logger.debug(
                f"[DOC_SCHEDULER] Adding new document processing job with interval {settings.interval_seconds}s"
            )
            self.scheduler.add_job(
                func=self._wrap_job(self._process_user_documents),
                args=[username],
                trigger="interval",
                seconds=settings.interval_seconds,
                id=job_id,
                name=f"Process Documents for {username}",
                jitter=30,  # Add small jitter to prevent multiple users from processing simultaneously
                max_instances=1,  # Prevent overlapping document processing for same user
                replace_existing=True,
            )

            session_info["scheduled_jobs"].add(job_id)
            logger.info(
                f"[DOC_SCHEDULER] Scheduled document processing job {job_id} for {username} with {settings.interval_seconds}s interval"
            )
            logger.debug(
                f"[DOC_SCHEDULER] User {username} now has {len(session_info['scheduled_jobs'])} scheduled jobs: {list(session_info['scheduled_jobs'])}"
            )

            # Verify job was added
            job = self.scheduler.get_job(job_id)
            if job:
                logger.info(
                    f"[DOC_SCHEDULER] Successfully verified job {job_id} exists, next run: {job.next_run_time}"
                )
            else:
                logger.error(
                    f"[DOC_SCHEDULER] Failed to verify job {job_id} exists!"
                )

            # Schedule (or tear down) the unindexed-document reconciler,
            # mirroring the document-processing job's lifecycle.
            self._schedule_reconciler(username, settings, session_info)

        except Exception as e:
            # No ``password`` local here, but the caller frame
            # (``_schedule_user_subscriptions``) holds the SQLCipher
            # master password — loguru ``diagnose=True`` walks the
            # frame stack and would render that caller-frame local.
            # Drop the traceback by using ``logger.warning`` without
            # ``exc_info``. ``redact_secrets`` with ``None`` is a no-op
            # here, but kept for the check-sensitive-logging pre-commit
            # hook + as a guide-post pairing for future refactors that
            # might bring a password into scope.
            safe_msg = redact_secrets(str(e), None)
            logger.warning(
                f"Error scheduling document processing for {username}: {safe_msg}"
            )

    def _schedule_reconciler(
        self,
        username: str,
        settings: DocumentSchedulerSettings,
        session_info: Dict[str, Any],
    ) -> None:
        """Add or remove the unindexed-document reconciler job.

        Mirrors the document-processing job lifecycle in
        ``_schedule_document_processing``: the job is (re)created only when
        EITHER ``sweep_library_collections`` OR ``generate_rag`` is enabled, and
        removed (and dropped from the session's tracked-jobs set) when both are
        off — so toggling the settings off and rescheduling tears the job down
        cleanly. The reconciler indexes every unindexed document (uploaded
        library docs AND research downloads), so both settings gate it: the
        ``generate_rag`` OR-arm preserves the legacy "index research downloads"
        behaviour that used to live inline in ``_process_user_documents``.
        """
        job_id = f"{username}_library_sweep"

        # Always remove any existing instance first so a disabled setting
        # tears the job down and a changed interval is re-applied.
        try:
            self.scheduler.remove_job(job_id)
            session_info["scheduled_jobs"].discard(job_id)
            logger.debug(f"[RECONCILER] Removed existing job {job_id}")
        except JobLookupError:
            pass  # Job doesn't exist, that's fine

        if not (settings.sweep_library_collections or settings.generate_rag):
            logger.debug(
                f"[RECONCILER] Indexing disabled for {username}; not scheduling"
            )
            return

        self.scheduler.add_job(
            func=self._wrap_job(self._reconcile_unindexed_documents),
            args=[username],
            trigger="interval",
            seconds=settings.interval_seconds,
            id=job_id,
            name=f"Unindexed Document Reconciler for {username}",
            jitter=60,
            max_instances=1,  # Self-rate-limit: no overlapping runs
            replace_existing=True,
        )
        session_info["scheduled_jobs"].add(job_id)
        logger.info(
            f"[RECONCILER] Scheduled unindexed-document reconciler job {job_id} "
            f"for {username} with {settings.interval_seconds}s interval"
        )

    def _arm_egress_backstop(self, settings_manager, username: str) -> None:
        """Set the audit-hook egress context from the user's saved settings so
        scheduled document downloads run under the same secondary net as an
        interactive research run. Best-effort and never raises — a backstop
        failure must not break the scheduler; the DownloadService PEP remains
        the primary gate. Cleared by the caller's @thread_cleanup on exit.
        """
        try:
            from ..security.egress.audit_hook import set_active_context
            from ..security.egress.policy import context_from_snapshot

            snapshot = settings_manager.get_settings_snapshot()
            if not isinstance(snapshot, dict):
                return
            primary = settings_manager.get_setting(
                "search.tool", DEFAULT_SEARCH_TOOL
            )
            ctx = context_from_snapshot(
                snapshot, primary or DEFAULT_SEARCH_TOOL, username=username
            )
            set_active_context(ctx)
        except Exception:
            logger.bind(policy_audit=True).debug(
                "doc scheduler: egress backstop not armed", exc_info=True
            )

    @thread_cleanup
    def _process_user_documents(self, username: str):
        """Process documents for a user."""
        logger.info(f"[DOC_SCHEDULER] Processing documents for user {username}")
        start_time = datetime.now(UTC)

        # Pre-declared so the except handlers can pass it to redact_secrets
        # even if the retrieve() call below itself raises.
        password = None
        try:
            session_info = self.user_sessions.get(username)
            if not session_info:
                logger.warning(
                    f"[DOC_SCHEDULER] No session info found for user {username}"
                )
                return

            password = self._credential_store.retrieve(username)
            if not password:
                logger.warning(
                    f"[DOC_SCHEDULER] Credentials expired for user {username}"
                )
                return
            logger.debug(
                f"[DOC_SCHEDULER] Starting document processing for {username}"
            )

            # Get user's document scheduler settings (cached)
            settings = self._get_document_scheduler_settings(username)

            logger.info(
                f"[DOC_SCHEDULER] Processing settings for {username}: "
                f"pdfs={settings.download_pdfs}, text={settings.extract_text}"
            )

            # RAG indexing has moved to ``_reconcile_unindexed_documents`` (its
            # own scheduled job), so ``generate_rag`` no longer drives any work
            # in this download/extract pass. Only the file-producing passes gate
            # whether this method runs.
            if not any(
                [
                    settings.download_pdfs,
                    settings.extract_text,
                ]
            ):
                logger.info(
                    f"[DOC_SCHEDULER] No download/extract options enabled for user {username}"
                )
                return

            # Parse last_run from cached settings
            last_run = (
                datetime.fromisoformat(settings.last_run)
                if settings.last_run
                else None
            )

            logger.info(f"[DOC_SCHEDULER] Last run for {username}: {last_run}")

            # Need database session for queries and updates
            from ..database.session_context import get_user_db_session
            from ..database.models.research import ResearchHistory
            from ..settings.manager import SettingsManager

            with get_user_db_session(username, password) as db:
                settings_manager = SettingsManager(db)

                # Arm the PEP-578 audit-hook backstop for this scheduled run.
                # The APScheduler worker thread carries no egress context, so
                # the secondary net would be inactive while DownloadService
                # fetches documents below. DownloadService's evaluate_url PEP
                # still gates each fetch (primary); this restores defense-in-
                # depth parity with an interactive run. @thread_cleanup clears
                # the context when this method returns.
                self._arm_egress_backstop(settings_manager, username)

                # Query for completed research since last run
                logger.debug(
                    f"[DOC_SCHEDULER] Querying for completed research since {last_run}"
                )
                query = db.query(ResearchHistory).filter(
                    ResearchHistory.status == ResearchStatus.COMPLETED,
                    ResearchHistory.completed_at.is_not(
                        None
                    ),  # Ensure completed_at is not null
                )

                if last_run:
                    query = query.filter(
                        ResearchHistory.completed_at > last_run
                    )

                # Limit to recent research to prevent overwhelming
                query = query.order_by(
                    ResearchHistory.completed_at.desc()
                ).limit(20)

                research_sessions = query.all()
                logger.debug(
                    f"[DOC_SCHEDULER] Query executed, found {len(research_sessions)} sessions"
                )

                if not research_sessions:
                    logger.info(
                        f"[DOC_SCHEDULER] No new completed research sessions found for user {username}"
                    )
                    return

                logger.info(
                    f"[DOC_SCHEDULER] Found {len(research_sessions)} research sessions to process for {username}"
                )

                # Log details of each research session
                for i, research in enumerate(
                    research_sessions[:5]
                ):  # Log first 5 details
                    title_safe = (
                        (research.title[:50] + "...")
                        if research.title
                        else "No title"
                    )
                    completed_safe = (
                        research.completed_at
                        if research.completed_at
                        else "No completion time"
                    )
                    logger.debug(
                        f"[DOC_SCHEDULER] Session {i + 1}: id={research.id}, title={title_safe}, completed={completed_safe}"
                    )

                    # Handle completed_at which might be a string or datetime
                    completed_at_obj = None
                    if research.completed_at:
                        if isinstance(research.completed_at, str):
                            try:
                                completed_at_obj = datetime.fromisoformat(
                                    research.completed_at.replace("Z", "+00:00")
                                )
                            except (ValueError, TypeError, AttributeError):
                                completed_at_obj = None
                        else:
                            completed_at_obj = research.completed_at

                    logger.debug(
                        f"[DOC_SCHEDULER]   - completed_at type: {type(research.completed_at)}"
                    )
                    logger.debug(
                        f"[DOC_SCHEDULER]   - completed_at timezone: {completed_at_obj.tzinfo if completed_at_obj else 'None'}"
                    )
                    logger.debug(f"[DOC_SCHEDULER]   - last_run: {last_run}")
                    logger.debug(
                        f"[DOC_SCHEDULER]   - completed_at > last_run: {completed_at_obj > last_run if last_run and completed_at_obj else 'N/A'}"
                    )

                # Capture a settings snapshot for this user/run so the
                # DownloadService below can build an EgressContext and
                # gate each per-resource URL. Without this the scheduler
                # would bypass policy entirely. Reuses the outer `db`
                # session (line 743) — get_settings_manager() in a
                # background thread must be passed a db_session
                # explicitly per the pre-commit thread-safety check.
                try:
                    user_settings_snapshot = (
                        settings_manager.get_settings_snapshot()
                    )
                except Exception as e:
                    # ``password`` is live in this frame (it opened the
                    # surrounding ``get_user_db_session``). Drop traceback
                    # + redact str(e) to avoid leaking the SQLCipher
                    # master password.
                    safe_msg = redact_secrets(str(e), password)
                    logger.warning(
                        f"[DOC_SCHEDULER] Could not build settings snapshot: "
                        f"{safe_msg} — downloads will not be scope-gated"
                    )
                    user_settings_snapshot = None

                processed_count = 0
                for research in research_sessions:
                    try:
                        logger.info(
                            f"[DOC_SCHEDULER] Processing research {research.id} for user {username}"
                        )

                        # Set search context so rate limiting works in both
                        # download_pdfs and extract_text paths
                        from ..utilities.thread_context import (
                            set_search_context,
                        )

                        set_search_context(
                            {
                                "research_id": str(research.id),
                                "username": username,
                                "user_password": password,
                                "research_phase": "document_scheduler",
                            }
                        )

                        # Call actual processing APIs
                        if settings.download_pdfs:
                            logger.info(
                                f"[DOC_SCHEDULER] Downloading PDFs for research {research.id}"
                            )
                            try:
                                # Use the DownloadService to queue PDF downloads
                                from ..research_library.services.download_service import (
                                    DownloadService,
                                )

                                with DownloadService(
                                    username,
                                    password,
                                    settings_snapshot=user_settings_snapshot,
                                ) as download_service:
                                    queued_count = download_service.queue_research_downloads(
                                        research.id
                                    )
                                    logger.info(
                                        f"[DOC_SCHEDULER] Queued {queued_count} PDF downloads for research {research.id}"
                                    )
                            except Exception as e:
                                # Recover the shared thread-local session
                                # before continuing — without rollback the
                                # next phase (text extract / RAG) and the
                                # post-loop last_run commit run on a
                                # poisoned session (issue #3827).
                                safe_rollback(db, "DOC_SCHEDULER PDF download")
                                # ``password`` is in scope and was passed
                                # into ``DownloadService``. Drop traceback
                                # + redact str(e) to avoid leaking the
                                # SQLCipher master password under
                                # ``diagnose=True``.
                                safe_msg = redact_secrets(str(e), password)
                                logger.warning(
                                    f"[DOC_SCHEDULER] Failed to download PDFs for research {research.id}: {safe_msg}"
                                )

                        if settings.extract_text:
                            logger.info(
                                f"[DOC_SCHEDULER] Extracting text for research {research.id}"
                            )
                            try:
                                # Use the DownloadService to extract text for all resources
                                from ..research_library.services.download_service import (
                                    DownloadService,
                                )
                                from ..database.models.research import (
                                    ResearchResource,
                                )

                                from ..research_library.utils import (
                                    is_downloadable_url,
                                )

                                with DownloadService(
                                    username,
                                    password,
                                    settings_snapshot=user_settings_snapshot,
                                ) as download_service:
                                    # Get all resources for this research (reuse existing db session)
                                    all_resources = (
                                        db.query(ResearchResource)
                                        .filter_by(research_id=research.id)
                                        .all()
                                    )
                                    # Filter: only process downloadable resources (academic/PDF)
                                    resources = [
                                        r
                                        for r in all_resources
                                        if is_downloadable_url(r.url)
                                    ]
                                    processed_count = 0
                                    for resource in resources:
                                        # We need to pass the password to the download service
                                        # The DownloadService creates its own database sessions, so we need to ensure password is available
                                        try:
                                            success, error = (
                                                download_service.download_as_text(
                                                    resource.id
                                                )
                                            )
                                            if success:
                                                processed_count += 1
                                                logger.info(
                                                    f"[DOC_SCHEDULER] Successfully extracted text for resource {resource.id}"
                                                )
                                            else:
                                                logger.warning(
                                                    f"[DOC_SCHEDULER] Failed to extract text for resource {resource.id}: {error}"
                                                )
                                        except Exception as resource_error:
                                            # Roll back FIRST so the next
                                            # iteration's queries don't
                                            # cascade on a poisoned session
                                            # (issue #3827).
                                            safe_rollback(
                                                db,
                                                "DOC_SCHEDULER resource",
                                            )
                                            # ``password`` is in scope and
                                            # was passed into the enclosing
                                            # ``DownloadService``. Drop the
                                            # traceback chain + redact str(e)
                                            # to avoid leaking the SQLCipher
                                            # master password.
                                            safe_msg = redact_secrets(
                                                str(resource_error), password
                                            )
                                            logger.warning(
                                                f"[DOC_SCHEDULER] Error processing resource {resource.id}: {safe_msg}"
                                            )
                                    logger.info(
                                        f"[DOC_SCHEDULER] Text extraction completed for research {research.id}: {processed_count}/{len(resources)} resources processed"
                                    )
                            except Exception as e:
                                safe_rollback(
                                    db, "DOC_SCHEDULER text extraction"
                                )
                                # ``password`` is in scope from the outer
                                # ``_process_user_documents`` retrieval —
                                # same redact + warning pattern as the
                                # inner handlers in this function.
                                safe_msg = redact_secrets(str(e), password)
                                logger.warning(
                                    f"[DOC_SCHEDULER] Failed to extract text for research {research.id}: {safe_msg}"
                                )

                        # NOTE: RAG indexing of research downloads used to live
                        # here (the old ``if settings.generate_rag:`` block).
                        # It has been retired — the unified
                        # ``_reconcile_unindexed_documents`` reconciler now
                        # indexes ALL unindexed documents (including research
                        # downloads that have no DocumentCollection row yet) on
                        # its own schedule, gated by ``generate_rag OR
                        # sweep_library_collections``. The download_pdfs and
                        # extract_text passes above remain here because they
                        # produce the ``text_content`` the reconciler indexes.

                        processed_count += 1
                        logger.debug(
                            f"[DOC_SCHEDULER] Successfully queued processing for research {research.id}"
                        )

                    except Exception as e:
                        safe_rollback(db, "DOC_SCHEDULER research")
                        # ``password`` is in scope from the outer
                        # ``_process_user_documents`` retrieval. Drop the
                        # traceback chain and redact str(e).
                        safe_msg = redact_secrets(str(e), password)
                        logger.warning(
                            f"[DOC_SCHEDULER] Error processing research {research.id} for user {username}: {safe_msg}"
                        )

                # Update last run time in user's settings.
                # Intentionally NOT wrapped in try/finally: if upstream setup
                # fails (DB open, SettingsManager init, initial query),
                # last_run should stay put so the next tick retries.
                # Advancing here would mask a persistent failure (corrupted
                # DB, wrong password). See closed PR #3288.
                current_time = datetime.now(UTC).isoformat()
                settings_manager.set_setting(
                    "document_scheduler.last_run", current_time, commit=True
                )
                logger.debug(
                    f"[DOC_SCHEDULER] Updated last run time for {username} to {current_time}"
                )

                end_time = datetime.now(UTC)
                duration = (end_time - start_time).total_seconds()
                logger.info(
                    f"[DOC_SCHEDULER] Completed document processing for user {username}: {processed_count} sessions processed in {duration:.2f}s"
                )

        except Exception as e:
            # ``password`` is pre-declared as ``None`` at the top of the
            # function, so it is always bound here even if the retrieve()
            # call itself raised. ``redact_secrets`` silently skips a
            # ``None`` secret. Drop the traceback chain.
            safe_msg = redact_secrets(str(e), password)
            logger.warning(
                f"[DOC_SCHEDULER] Error processing documents for user {username}: {safe_msg}"
            )

    @thread_cleanup
    def _reconcile_unindexed_documents(self, username: str) -> None:
        """Unified background reconciler that indexes ANY unindexed document.

        Self-healing follow-up to the immediate auto-index queue (PR #3939),
        which caps the queue and DROPS documents on saturation, AND replacement
        for the retired research-scoped ``generate_rag`` indexing block that
        used to live inline in ``_process_user_documents``. A single scheduled
        job now covers every unindexed document, so library uploads and research
        downloads can no longer be permanently missed.

        Two cases are handled per tick, each with its OWN independent
        ``_LIBRARY_SWEEP_BATCH`` budget so the work done in a single thread tick
        stays bounded (total <= 2 x ``_LIBRARY_SWEEP_BATCH``). The budgets are
        decoupled on purpose: case (a) only marks a row indexed on SUCCESS, so a
        block of permanently-failing case-(a) rows must not be able to consume
        case (b)'s budget and starve the research-orphan path:

        (a) In-collection unindexed: documents that already have a
            ``DocumentCollection`` link (e.g. manual uploads — ``upload_to_
            collection`` always creates the row) with ``indexed`` False and
            text content. Indexed via the per-collection RAG factory so each
            collection's own embedding config is honored.
        (b) Research orphans: ``Document`` rows with ``research_id`` set and
            text content that have NO ``DocumentCollection`` link in the default
            library collection yet (research downloads that were never
            ingested). ``index_document(doc_id, default_library_id, ...)`` calls
            ``ensure_in_collection`` internally, so it ingests + indexes in one
            call.

        Behaviour:
        - Gated by EITHER ``sweep_library_collections`` OR ``generate_rag`` —
          the ``generate_rag`` arm preserves the legacy "index research
          downloads" behaviour. Off by default; early-returns when neither set.
        - Idempotent: uses ``index_document(..., force_reindex=False)`` and only
          selects rows that are not yet indexed, so already-indexed documents
          are never touched.
        - Self-rate-limited: each case is capped at ``_LIBRARY_SWEEP_BATCH``
          documents per tick (so total <= 2 x ``_LIBRARY_SWEEP_BATCH``); the job
          is scheduled with ``max_instances=1``.
        """
        logger.info(
            f"[RECONCILER] Starting unindexed-document reconcile for user {username}"
        )

        # Pre-declared so the except handlers can pass it to redact_secrets
        # even if the retrieve() call below itself raises.
        password = None
        try:
            session_info = self.user_sessions.get(username)
            if not session_info:
                logger.warning(
                    f"[RECONCILER] No session info found for user {username}"
                )
                return

            password = self._credential_store.retrieve(username)
            if not password:
                logger.warning(
                    f"[RECONCILER] Credentials expired for user {username}"
                )
                return

            # Get user's document scheduler settings (cached).
            settings = self._get_document_scheduler_settings(username)

            # Gate at runtime too, not just at scheduling: the already-live
            # APScheduler job keeps firing after the document scheduler is
            # disabled until the next reschedule, and the setting description
            # promises the sweep only runs while the scheduler is enabled.
            # OFF by default: runs only when the scheduler is enabled AND
            # EITHER opt-in is set. The generate_rag arm preserves the legacy
            # research-download indexing behaviour from _process_user_documents.
            if not settings.enabled or not (
                settings.sweep_library_collections or settings.generate_rag
            ):
                logger.debug(
                    f"[RECONCILER] Indexing disabled for user {username}"
                )
                return

            # Lazy import of the RAG factory. Imported here (not at module
            # top) to keep the import surface of this scheduler module small
            # and consistent with the other lazy imports in this file; the
            # factory itself has no import dependency on the scheduler so a
            # top-level import would also be safe.
            from ..research_library.services.rag_service_factory import (
                get_rag_service,
            )

            from ..database.session_context import get_user_db_session
            from ..settings.manager import SettingsManager

            with get_user_db_session(username, password) as db:
                settings_manager = SettingsManager(db)

                # Arm the PEP-578 audit-hook backstop for this scheduled run,
                # mirroring _process_user_documents. Indexing itself doesn't
                # download, but embedding providers may make network calls;
                # this keeps defense-in-depth parity with an interactive run.
                # Cleared by @thread_cleanup on exit.
                self._arm_egress_backstop(settings_manager, username)

                total_indexed = 0

                # ---- Case (a): in-collection unindexed documents ----------
                # Bounded by _LIBRARY_SWEEP_BATCH so a large backlog self-heals
                # over successive ticks. Only rows with actual text content can
                # be indexed. RANDOMIZED selection (not a stable id order): a
                # row leaves this candidate set only on SUCCESS and we track no
                # per-row failure state, so a deterministic order would let a
                # block of permanently-failing low-id rows (e.g. empty-text /
                # scanned PDFs that always return an indexing error yet pass the
                # text_content IS NOT NULL filter) win the LIMIT slots every
                # tick and starve indexable higher-id rows forever. Random
                # sampling gives every indexable row a chance each tick, so
                # progress is eventually made despite a permanent-failure set.
                unindexed = (
                    db.query(
                        DocumentCollection.document_id,
                        DocumentCollection.collection_id,
                    )
                    .join(
                        Document,
                        Document.id == DocumentCollection.document_id,
                    )
                    .filter(
                        DocumentCollection.indexed.is_(False),
                        Document.text_content.isnot(None),
                    )
                    .order_by(func.random())
                    .limit(_LIBRARY_SWEEP_BATCH)
                    .all()
                )

                # Group document ids by collection so we build exactly one RAG
                # service per collection (each collection can have its own
                # embedding config).
                docs_by_collection: Dict[str, List[str]] = {}
                for doc_id, coll_id in unindexed:
                    docs_by_collection.setdefault(coll_id, []).append(doc_id)

                if unindexed:
                    logger.info(
                        f"[RECONCILER] Found {len(unindexed)} in-collection "
                        f"unindexed document(s) across "
                        f"{len(docs_by_collection)} collection(s) for {username}"
                    )

                for coll_id, doc_ids in docs_by_collection.items():
                    try:
                        # USE THE FACTORY so per-collection embedding settings
                        # (model/provider/chunking/etc.) stored on the
                        # collection are honored — get_rag_service loads them
                        # from the collection row when collection_id is given.
                        with get_rag_service(
                            username,
                            collection_id=coll_id,
                            db_password=password,
                        ) as rag_service:
                            for doc_id in doc_ids:
                                try:
                                    result = rag_service.index_document(
                                        document_id=doc_id,
                                        collection_id=coll_id,
                                        force_reindex=False,
                                    )
                                    if result.get("status") == "success":
                                        total_indexed += 1
                                        logger.debug(
                                            f"[RECONCILER] Indexed document {doc_id} "
                                            f"into collection {coll_id} with "
                                            f"{result.get('chunk_count', 0)} chunks"
                                        )
                                except Exception as doc_error:
                                    # ``password`` is in scope and was passed
                                    # into ``get_rag_service``. Drop the
                                    # traceback chain + redact str(e) to avoid
                                    # leaking the SQLCipher master password.
                                    safe_msg = redact_secrets(
                                        str(doc_error), password
                                    )
                                    logger.warning(
                                        f"[RECONCILER] Failed to index document "
                                        f"{doc_id} into collection {coll_id}: {safe_msg}"
                                    )
                    except Exception as coll_error:
                        # Recover the shared thread-local session before moving
                        # on to the next collection so its queries don't run on
                        # a poisoned session.
                        safe_rollback(db, "RECONCILER collection")
                        safe_msg = redact_secrets(str(coll_error), password)
                        logger.warning(
                            f"[RECONCILER] Failed to index collection {coll_id}: {safe_msg}"
                        )

                # ---- Case (b): research orphans -> default library ---------
                # Research downloads land as Document rows (research_id set)
                # with NO DocumentCollection link yet. index_document()
                # ensure_in_collection's the default-library link, so this
                # ingests + indexes in one call.
                #
                # Case (b) gets its OWN independent _LIBRARY_SWEEP_BATCH budget
                # rather than the leftover of case (a). Case (a) only flips a
                # row to indexed=True on SUCCESS, so a block of permanently
                # failing case-(a) rows (empty text, embedding/FAISS errors,
                # PolicyDeniedError under egress denial) would otherwise fill
                # the LIMIT every tick, leave the leftover at 0, and starve the
                # research-orphan path forever — regressing the no-regression
                # promise for generate_rag-only users. Decoupling the budgets
                # caps total work at 2 x _LIBRARY_SWEEP_BATCH per tick, which is
                # acceptable. RANDOMIZED selection (see case (a)) so a block of
                # permanently-failing low-id orphans can't pin the LIMIT slots
                # every tick and starve the rest.
                #
                # Resolve the default library collection once.
                default_library_id = get_default_library_id(username, password)

                orphans = (
                    db.query(Document.id)
                    .outerjoin(
                        DocumentCollection,
                        (DocumentCollection.document_id == Document.id)
                        & (
                            DocumentCollection.collection_id
                            == default_library_id
                        ),
                    )
                    .filter(
                        Document.research_id.isnot(None),
                        Document.text_content.isnot(None),
                        DocumentCollection.id.is_(None),
                    )
                    .order_by(func.random())
                    .limit(_LIBRARY_SWEEP_BATCH)
                    .all()
                )

                if orphans:
                    logger.info(
                        f"[RECONCILER] Found {len(orphans)} research "
                        f"orphan document(s) to ingest into the default "
                        f"library for {username}"
                    )
                    try:
                        with get_rag_service(
                            username,
                            collection_id=default_library_id,
                            db_password=password,
                        ) as rag_service:
                            for (doc_id,) in orphans:
                                try:
                                    result = rag_service.index_document(
                                        document_id=doc_id,
                                        collection_id=default_library_id,
                                        force_reindex=False,
                                    )
                                    if result.get("status") == "success":
                                        total_indexed += 1
                                        logger.debug(
                                            f"[RECONCILER] Ingested + "
                                            f"indexed research orphan "
                                            f"{doc_id} into the default "
                                            f"library with "
                                            f"{result.get('chunk_count', 0)} chunks"
                                        )
                                except Exception as doc_error:
                                    safe_msg = redact_secrets(
                                        str(doc_error), password
                                    )
                                    logger.warning(
                                        f"[RECONCILER] Failed to index "
                                        f"research orphan {doc_id}: {safe_msg}"
                                    )
                    except Exception as orphan_error:
                        safe_rollback(db, "RECONCILER orphans")
                        safe_msg = redact_secrets(str(orphan_error), password)
                        logger.warning(
                            f"[RECONCILER] Failed to index research "
                            f"orphans into the default library: {safe_msg}"
                        )

                logger.info(
                    f"[RECONCILER] Completed reconcile for user {username}: "
                    f"{total_indexed} document(s) indexed "
                    f"(per-case batch cap {_LIBRARY_SWEEP_BATCH})"
                )

        except Exception as e:
            # ``password`` is pre-declared as ``None`` at the top of the
            # function, so it is always bound here even if the retrieve()
            # call itself raised. ``redact_secrets`` silently skips a
            # ``None`` secret. Drop the traceback chain.
            safe_msg = redact_secrets(str(e), password)
            logger.warning(
                f"[RECONCILER] Error reconciling unindexed documents for user {username}: {safe_msg}"
            )

    def get_document_scheduler_status(self, username: str) -> Dict[str, Any]:
        """Get document scheduler status for a specific user."""
        try:
            session_info = self.user_sessions.get(username)
            if not session_info:
                return {
                    "enabled": False,
                    "message": "User not found in scheduler",
                }

            # Get user's document scheduler settings (cached)
            settings = self._get_document_scheduler_settings(username)

            # Check if user has document processing job
            job_id = f"{username}_document_processing"
            has_job = job_id in session_info.get("scheduled_jobs", set())

            return {
                "enabled": settings.enabled,
                "interval_seconds": settings.interval_seconds,
                "processing_options": {
                    "download_pdfs": settings.download_pdfs,
                    "extract_text": settings.extract_text,
                    # generate_rag and sweep_library_collections both gate the
                    # unified reconciler (_reconcile_unindexed_documents).
                    "generate_rag": settings.generate_rag,
                    "sweep_library_collections": settings.sweep_library_collections,
                },
                "last_run": settings.last_run,
                "has_scheduled_job": has_job,
                "user_active": username in self.user_sessions,
            }

        except Exception as e:
            # No ``password`` local in this method, but caller frames
            # (e.g. a route handler that already retrieved the user's
            # password) could be rendered under loguru ``diagnose=True``.
            # Drop the traceback by using ``logger.warning`` without
            # ``exc_info``.
            safe_msg = redact_secrets(str(e), None)
            logger.warning(
                f"Error getting document scheduler status for user {username}: {safe_msg}"
            )
            return {
                "enabled": False,
                "message": f"Failed to retrieve scheduler status: {type(e).__name__}",
            }

    def trigger_document_processing(self, username: str) -> bool:
        """Trigger immediate document processing for a user."""
        logger.info(
            f"[DOC_SCHEDULER] Manual trigger requested for user {username}"
        )
        try:
            session_info = self.user_sessions.get(username)
            if not session_info:
                logger.warning(
                    f"[DOC_SCHEDULER] User {username} not found in scheduler"
                )
                logger.debug(
                    f"[DOC_SCHEDULER] Available users: {list(self.user_sessions.keys())}"
                )
                return False

            if not self.is_running:
                logger.warning(
                    f"[DOC_SCHEDULER] Scheduler not running, cannot trigger document processing for {username}"
                )
                return False

            # Trigger immediate processing
            job_id = f"{username}_document_processing_manual"
            logger.debug(f"[DOC_SCHEDULER] Scheduling manual job {job_id}")

            self.scheduler.add_job(
                func=self._wrap_job(self._process_user_documents),
                args=[username],
                trigger="date",
                run_date=datetime.now(UTC) + timedelta(seconds=1),
                id=job_id,
                name=f"Manual Document Processing for {username}",
                replace_existing=True,
            )

            # Verify job was added
            job = self.scheduler.get_job(job_id)
            if job:
                logger.info(
                    f"[DOC_SCHEDULER] Successfully triggered manual document processing for user {username}, job {job_id}, next run: {job.next_run_time}"
                )
            else:
                logger.error(
                    f"[DOC_SCHEDULER] Failed to verify manual job {job_id} was added!"
                )
                return False

            return True

        except Exception as e:
            # No ``password`` local in this method, but caller frames
            # could hold one — drop the traceback to avoid frame-local
            # rendering under ``diagnose=True``.
            safe_msg = redact_secrets(str(e), None)
            logger.warning(
                f"[DOC_SCHEDULER] Error triggering document processing for user {username}: {safe_msg}"
            )
            return False

    @thread_cleanup
    def _check_user_overdue_subscriptions(self, username: str):
        """Check and immediately run any overdue subscriptions for a user."""
        # Pre-declared so the except handler can pass it to redact_secrets
        # even if the retrieve() call below itself raises.
        password = None
        try:
            session_info = self.user_sessions.get(username)
            if not session_info:
                return

            password = self._credential_store.retrieve(username)
            if not password:
                return

            # Get user's overdue subscriptions
            from ..database.session_context import get_user_db_session
            from ..database.models.news import NewsSubscription
            from datetime import timezone

            with get_user_db_session(username, password) as db:
                now = datetime.now(timezone.utc)
                overdue_subs = (
                    db.query(NewsSubscription)
                    .filter(NewsSubscription.due_filter(now))
                    .all()
                )

            if overdue_subs:
                logger.info(
                    f"Found {len(overdue_subs)} overdue subscriptions for {username}"
                )

                for sub in overdue_subs:
                    # Run immediately with small random delay
                    # Security: random delay to stagger overdue jobs, not security-sensitive
                    delay_seconds = random.randint(1, 30)
                    job_id = (
                        f"overdue_{username}_{sub.id}_{int(now.timestamp())}"
                    )

                    self.scheduler.add_job(
                        func=self._wrap_job(self._check_subscription),
                        args=[username, sub.id],
                        trigger="date",
                        run_date=now + timedelta(seconds=delay_seconds),
                        id=job_id,
                        name=f"Overdue: {sub.name or sub.query_or_topic[:30]}",
                        replace_existing=True,
                    )

                    logger.info(
                        f"Scheduled overdue subscription {sub.id} to run in {delay_seconds} seconds"
                    )

        except Exception as e:
            # ``password`` was retrieved above and passed into
            # ``get_user_db_session``. Drop traceback + redact str(e)
            # to avoid leaking the SQLCipher master password.
            safe_msg = redact_secrets(str(e), password)
            logger.warning(
                f"Error checking overdue subscriptions for {username}: {safe_msg}"
            )

    @thread_cleanup
    def _check_subscription(self, username: str, subscription_id: int):
        """Check and refresh a single subscription."""
        logger.info(
            f"_check_subscription called for user {username}, subscription {subscription_id}"
        )
        # Pre-declared so the except handler can pass it to redact_secrets
        # even if the retrieve() call below itself raises.
        password = None
        try:
            session_info = self.user_sessions.get(username)
            if not session_info:
                # User no longer active, cancel job
                job_id = f"{username}_{subscription_id}"
                try:
                    self.scheduler.remove_job(job_id)
                except JobLookupError:
                    pass
                return

            password = self._credential_store.retrieve(username)
            if not password:
                logger.warning(
                    f"Credentials expired for {username}, skipping subscription check"
                )
                return

            # Get subscription details
            from ..database.session_context import get_user_db_session
            from ..database.models.news import (
                NewsSubscription,
                SubscriptionStatus,
            )
            from ..news.subscription_runner import advance_refresh_schedule

            with get_user_db_session(username, password) as db:
                sub = db.query(NewsSubscription).get(subscription_id)
                if not sub or sub.status != SubscriptionStatus.ACTIVE.value:
                    logger.info(
                        f"Subscription {subscription_id} not active, skipping"
                    )
                    return

                # Prepare query with date replacement using user's timezone
                query = sub.query_or_topic
                if "YYYY-MM-DD" in query:
                    from local_deep_research.news.core.utils import (
                        get_local_date_string,
                    )
                    from ..settings.manager import SettingsManager

                    settings_manager = SettingsManager(db)
                    local_date = get_local_date_string(settings_manager)
                    query = query.replace("YYYY-MM-DD", local_date)

                # Update last/next refresh times
                advance_refresh_schedule(sub, datetime.now(UTC))
                db.commit()

                subscription_data = {
                    "id": sub.id,
                    "name": sub.name,
                    "query": query,
                    "original_query": sub.query_or_topic,
                    "model_provider": sub.model_provider,
                    "model": sub.model,
                    "search_strategy": sub.search_strategy,
                    "search_engine": sub.search_engine,
                }

            logger.info(
                f"Refreshing subscription {subscription_id}: {subscription_data['name']}"
            )

            # Trigger research synchronously using requests with proper auth
            self._trigger_subscription_research_sync(
                username, subscription_data
            )

            # Reschedule for next interval if using interval trigger
            job_id = f"{username}_{subscription_id}"
            job = self.scheduler.get_job(job_id)
            if job and job.trigger.__class__.__name__ == "DateTrigger":
                # For date triggers, reschedule
                # Security: random jitter to distribute subscription timing, not security-sensitive
                next_run = datetime.now(UTC) + timedelta(
                    minutes=sub.refresh_interval_minutes,
                    seconds=random.randint(
                        0, int(self.config.get("max_jitter_seconds", 300))
                    ),
                )
                self.scheduler.add_job(
                    func=self._wrap_job(self._check_subscription),
                    args=[username, subscription_id],
                    trigger="date",
                    run_date=next_run,
                    id=job_id,
                    replace_existing=True,
                )

        except Exception as e:
            # ``password`` was retrieved above and passed into
            # ``get_user_db_session``. Drop traceback + redact str(e)
            # to avoid leaking the SQLCipher master password.
            safe_msg = redact_secrets(str(e), password)
            logger.warning(
                f"Error checking subscription {subscription_id}: {safe_msg}"
            )

    @thread_cleanup
    def _trigger_subscription_research_sync(
        self, username: str, subscription: Dict[str, Any]
    ):
        """Trigger research for a subscription using programmatic API."""
        from ..config.thread_settings import set_settings_context

        # Pre-declared so the except handler can pass it to redact_secrets
        # even if the retrieve() call below itself raises.
        password = None
        try:
            # Get user's password from session info
            session_info = self.user_sessions.get(username)
            if not session_info:
                logger.error(f"No session info for user {username}")
                return

            password = self._credential_store.retrieve(username)
            if not password:
                logger.error(f"Credentials expired for user {username}")
                return

            # Generate research ID
            import uuid

            research_id = str(uuid.uuid4())

            logger.info(
                f"Starting research {research_id} for subscription {subscription['id']}"
            )

            # Get user settings for research
            from ..database.session_context import get_user_db_session
            from ..settings.manager import SettingsManager

            with get_user_db_session(username, password) as db:
                settings_manager = SettingsManager(db)
                settings_snapshot = settings_manager.get_settings_snapshot()

                # Use the search engine from the subscription if specified
                search_engine = subscription.get("search_engine")

                if search_engine:
                    settings_snapshot["search.tool"] = {
                        "value": search_engine,
                        "ui_element": "select",
                    }
                    logger.info(
                        f"Using subscription's search engine: '{search_engine}' for {subscription['id']}"
                    )
                else:
                    # Use the user's default search tool from their settings
                    default_search_tool = settings_snapshot.get(
                        "search.tool", DEFAULT_SEARCH_TOOL
                    )
                    logger.info(
                        f"Using user's default search tool: '{default_search_tool}' for {subscription['id']}"
                    )

                logger.debug(
                    f"Settings snapshot has {len(settings_snapshot)} settings"
                )
                # Log a few key settings to verify they're present
                logger.debug(
                    f"Key settings: llm.model={settings_snapshot.get('llm.model')}, llm.provider={settings_snapshot.get('llm.provider')}, search.tool={settings_snapshot.get('search.tool')}"
                )

            # Set up research parameters
            query = subscription["query"]

            # Build metadata for news search
            metadata = {
                "is_news_search": True,
                "search_type": "news_analysis",
                "display_in": "news_feed",
                "subscription_id": subscription["id"],
                "triggered_by": "scheduler",
                "subscription_name": subscription["name"],
                "title": subscription["name"] if subscription["name"] else None,
                "scheduled_at": datetime.now(UTC).isoformat(),
                "original_query": subscription["original_query"],
                "user_id": username,
            }

            # Use programmatic API with settings context
            from ..api.research_functions import quick_summary

            # Create and set settings context for this thread
            settings_context = SnapshotSettingsContext(settings_snapshot)
            set_settings_context(settings_context)

            # Get search strategy from subscription data
            search_strategy = subscription.get("search_strategy")

            # Build kwargs for quick_summary, only including
            # search_strategy if the subscription specifies one.
            quick_summary_kwargs = {
                "query": query,
                "research_id": research_id,
                "username": username,
                "user_password": password,
                "settings_snapshot": settings_snapshot,
                "model_name": subscription.get("model"),
                "provider": subscription.get("model_provider"),
                "metadata": metadata,
                "search_original_query": False,  # Don't send long subscription prompts to search engines
            }
            if search_strategy:
                quick_summary_kwargs["search_strategy"] = search_strategy

            result = quick_summary(**quick_summary_kwargs)

            logger.info(
                f"Completed research {research_id} for subscription {subscription['id']}"
            )

            # Store the research result in the database
            self._store_research_result(
                username,
                password,
                research_id,
                subscription["id"],
                result,
                subscription,
            )

        except Exception as e:
            # ``password`` was retrieved from the credential store at
            # the top of this function and passed through to
            # ``get_user_db_session``, ``quick_summary``
            # (``user_password``), and ``_store_research_result``. A
            # SQLAlchemy / requests exception from any of those paths
            # could carry frame locals containing the SQLCipher master
            # password — drop the traceback chain and redact str(e).
            safe_msg = redact_secrets(str(e), password)
            logger.warning(
                f"Error triggering research for subscription {subscription['id']}: {safe_msg}"
            )

    def _store_research_result(
        self,
        username: str,
        password: str,
        research_id: str,
        subscription_id: int,
        result: Dict[str, Any],
        subscription: Dict[str, Any],
    ):
        """Store research result in database for news display."""
        try:
            from ..database.session_context import get_user_db_session
            from ..database.models import ResearchHistory
            from ..settings.manager import SettingsManager
            import json

            # Convert result to JSON-serializable format
            def make_serializable(obj):
                """Convert non-serializable objects to dictionaries."""
                if hasattr(obj, "dict"):
                    return obj.dict()
                if hasattr(obj, "__dict__"):
                    return {
                        k: make_serializable(v)
                        for k, v in obj.__dict__.items()
                        if not k.startswith("_")
                    }
                if isinstance(obj, (list, tuple)):
                    return [make_serializable(item) for item in obj]
                if isinstance(obj, dict):
                    return {k: make_serializable(v) for k, v in obj.items()}
                return obj

            serializable_result = make_serializable(result)

            with get_user_db_session(username, password) as db:
                # Get user settings to store in metadata
                settings_manager = SettingsManager(db)
                settings_snapshot = settings_manager.get_settings_snapshot()

                # Get the report content - check both 'report' and 'summary' fields
                report_content = serializable_result.get(
                    "report"
                ) or serializable_result.get("summary")
                logger.debug(
                    f"Report content length: {len(report_content) if report_content else 0} chars"
                )

                # Extract sources/links from the result. They get
                # persisted to research_resources AFTER history_entry
                # commits below (FK requires research_id to exist).
                sources = serializable_result.get("sources", [])

                # Then format citations in the report content
                if report_content:
                    # Import citation formatter
                    from ..text_optimization.citation_formatter import (
                        CitationFormatter,
                        CitationMode,
                    )
                    from ..config.search_config import (
                        get_setting_from_snapshot,
                    )

                    # Get citation format from settings
                    citation_format = get_setting_from_snapshot(
                        "report.citation_format", "domain_id_hyperlinks"
                    )
                    mode_map = {
                        "number_hyperlinks": CitationMode.NUMBER_HYPERLINKS,
                        "domain_hyperlinks": CitationMode.DOMAIN_HYPERLINKS,
                        "domain_id_hyperlinks": CitationMode.DOMAIN_ID_HYPERLINKS,
                        "domain_id_always_hyperlinks": CitationMode.DOMAIN_ID_ALWAYS_HYPERLINKS,
                        "source_tagged_hyperlinks": CitationMode.SOURCE_TAGGED_HYPERLINKS,
                        "no_hyperlinks": CitationMode.NO_HYPERLINKS,
                    }
                    mode = mode_map.get(
                        citation_format, CitationMode.DOMAIN_ID_HYPERLINKS
                    )
                    formatter = CitationFormatter(mode=mode)

                    # Format citations within the content
                    report_content = formatter.format_document(report_content)

                if not report_content:
                    # If neither field exists, use the full result as JSON
                    report_content = json.dumps(serializable_result)

                # Generate headline and topics for news searches
                from ..news.utils.headline_generator import generate_headline
                from ..news.utils.topic_generator import generate_topics

                query_text = result.get(
                    "query", subscription.get("query", "News Update")
                )

                # Generate headline from the actual research findings
                logger.info(
                    f"Generating headline for subscription {subscription_id}"
                )
                generated_headline = generate_headline(
                    query=query_text,
                    findings=report_content,
                    max_length=200,  # Allow longer headlines for news
                    settings_snapshot=settings_snapshot,
                )

                # Generate topics from the findings
                logger.info(
                    f"Generating topics for subscription {subscription_id}"
                )
                generated_topics = generate_topics(
                    query=query_text,
                    findings=report_content,
                    category=subscription.get("name", "News"),
                    max_topics=6,
                    settings_snapshot=settings_snapshot,
                )

                logger.info(
                    f"Generated headline: {generated_headline}, topics: {generated_topics}"
                )

                # Get subscription name for metadata
                subscription_name = subscription.get("name", "")

                # Use generated headline as title, or fallback
                if generated_headline:
                    title = generated_headline
                else:
                    if subscription_name:
                        title = f"{subscription_name} - {datetime.now(UTC).isoformat(timespec='minutes')}"
                    else:
                        title = f"{query_text[:60]}... - {datetime.now(UTC).isoformat(timespec='minutes')}"

                # Create research history entry
                history_entry = ResearchHistory(
                    id=research_id,
                    query=result.get("query", ""),
                    mode="news_subscription",
                    status="completed",
                    created_at=datetime.now(UTC).isoformat(),
                    completed_at=datetime.now(UTC).isoformat(),
                    title=title,
                    research_meta={
                        "subscription_id": subscription_id,
                        "triggered_by": "scheduler",
                        "is_news_search": True,
                        "username": username,
                        "subscription_name": subscription_name,  # Store subscription name for display
                        "settings_snapshot": settings_snapshot,  # Store settings snapshot for later retrieval
                        "generated_headline": generated_headline,  # Store generated headline for news display
                        "generated_topics": generated_topics,  # Store topics for categorization
                    },
                )
                db.add(history_entry)
                db.commit()

                # Persist sources to research_resources so the assembler
                # can rebuild the Sources block at render time. Was
                # previously written INLINE into report_content via a
                # "## Sources" tail — the report_content refactor moves
                # this to structured storage matching normal research.
                if sources:
                    try:
                        from ..web.services.research_sources_service import (
                            ResearchSourcesService,
                        )

                        ResearchSourcesService.save_research_sources(
                            research_id=research_id,
                            sources=sources,
                            username=username,
                        )
                    except Exception as e:
                        # ``password`` is a parameter of this method —
                        # don't render a traceback that could expose it
                        # via diagnose=True frame locals.
                        safe_msg = redact_secrets(str(e), password)
                        logger.warning(
                            "Failed to persist scheduler sources for "
                            "research {} — assembler will render no Sources "
                            "block for this row: {}",
                            research_id,
                            safe_msg,
                        )

                # Store the report content using storage abstraction
                from ..storage import get_report_storage

                # Use storage to save the report (report_content already retrieved above)
                storage = get_report_storage(session=db)
                storage.save_report(
                    research_id=research_id,
                    content=report_content,
                    username=username,
                )

                logger.info(
                    f"Stored research result {research_id} for subscription {subscription_id}"
                )

        except Exception as e:
            # ``password`` is a function parameter, so it is always in
            # this frame. Drop traceback + redact str(e) to avoid leaking
            # the SQLCipher master password.
            safe_msg = redact_secrets(str(e), password)
            logger.warning(f"Error storing research result: {safe_msg}")

    def _run_cleanup_with_tracking(self):
        """Wrapper that tracks cleanup execution."""

        try:
            cleaned_count = self._cleanup_inactive_users()

            logger.info(
                f"Cleanup successful: removed {cleaned_count} inactive users"
            )

        except Exception:
            logger.exception("Cleanup job failed")

    def _cleanup_inactive_users(self) -> int:
        """Remove users inactive for longer than retention period."""
        retention_hours = self.config.get("retention_hours", 48)
        cutoff = datetime.now(UTC) - timedelta(hours=retention_hours)

        cleaned_count = 0

        with self.lock:
            inactive_users = [
                user_id
                for user_id, session in self.user_sessions.items()
                if session["last_activity"] < cutoff
            ]

            for user_id in inactive_users:
                # Remove all scheduled jobs
                for job_id in self.user_sessions[user_id][
                    "scheduled_jobs"
                ].copy():
                    try:
                        self.scheduler.remove_job(job_id)
                    except JobLookupError:
                        pass

                # Clear credentials and session data
                self._credential_store.clear(user_id)
                del self.user_sessions[user_id]
                cleaned_count += 1
                logger.info(f"Cleaned up inactive user {user_id}")

        return cleaned_count

    def _reload_config(self):
        """Reload configuration from settings manager."""
        if not hasattr(self, "settings_manager") or not self.settings_manager:
            return

        try:
            old_retention = self.config.get("retention_hours", 48)

            # Reload all settings
            for key in self.config:
                if key == "enabled":
                    continue  # Don't change enabled state while running

                full_key = f"news.scheduler.{key}"
                self.config[key] = self._get_setting(full_key, self.config[key])

            # Handle changes that need immediate action
            if old_retention != self.config["retention_hours"]:
                logger.info(
                    f"Retention period changed from {old_retention} "
                    f"to {self.config['retention_hours']} hours"
                )
                # Trigger immediate cleanup with new retention
                self.scheduler.add_job(
                    self._wrap_job(self._run_cleanup_with_tracking),
                    "date",
                    run_date=datetime.now(UTC) + timedelta(seconds=5),
                    id="immediate_cleanup_config_change",
                )

            # Clear settings cache to pick up any user setting changes
            self.invalidate_all_settings_cache()

        except Exception:
            logger.exception("Error reloading configuration")

    def get_status(self) -> Dict[str, Any]:
        """Get scheduler status information."""
        with self.lock:
            active_users = len(self.user_sessions)
            total_jobs = sum(
                len(session["scheduled_jobs"])
                for session in self.user_sessions.values()
            )

        # Get next run time for cleanup job
        next_cleanup = None
        if self.is_running:
            job = self.scheduler.get_job("cleanup_inactive_users")
            if job:
                next_cleanup = job.next_run_time

        return {
            "is_running": self.is_running,
            "config": self.config,
            "active_users": active_users,
            "total_scheduled_jobs": total_jobs,
            "next_cleanup": next_cleanup.isoformat() if next_cleanup else None,
            "memory_usage": self._estimate_memory_usage(),
        }

    def _estimate_memory_usage(self) -> int:
        """Estimate memory usage of user sessions."""

        # Rough estimate: username (50) + password (100) + metadata (200) per user
        per_user_estimate = 350
        return len(self.user_sessions) * per_user_estimate

    def get_user_sessions_summary(self) -> List[Dict[str, Any]]:
        """Get summary of active user sessions (without passwords)."""
        with self.lock:
            summary = []
            for user_id, session in self.user_sessions.items():
                summary.append(
                    {
                        "user_id": user_id,
                        "last_activity": session["last_activity"].isoformat(),
                        "scheduled_jobs": len(session["scheduled_jobs"]),
                        "time_since_activity": str(
                            datetime.now(UTC) - session["last_activity"]
                        ),
                    }
                )
            return summary


# Singleton instance getter
_scheduler_instance = None


def get_background_job_scheduler() -> BackgroundJobScheduler:
    """Get the singleton news scheduler instance."""
    global _scheduler_instance
    if _scheduler_instance is None:
        _scheduler_instance = BackgroundJobScheduler()
    return _scheduler_instance
