Coverage for src/local_deep_research/scheduler/background.py: 94%
869 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-06 15:42 +0000
1"""
2Activity-based news subscription scheduler for per-user encrypted databases.
3Tracks user activity and temporarily stores credentials for automatic updates.
4"""
6import random
7import threading
8from dataclasses import dataclass
9from datetime import datetime, timedelta, UTC
10from functools import wraps
11from typing import Any, Callable, Dict, List
13from cachetools import TTLCache
14from loguru import logger
15from ..settings.logger import log_settings
16from ..settings.manager import SnapshotSettingsContext
18from apscheduler.schedulers.background import BackgroundScheduler
19from apscheduler.jobstores.base import JobLookupError
20from sqlalchemy import func
21from ..constants import ResearchStatus
22from ..database.credential_store_base import CredentialStoreBase
23from ..database.session_context import safe_rollback
24from ..database.thread_local_session import thread_cleanup
25from ..security.log_sanitizer import redact_secrets
27# RAG indexing imports. The reconciler builds RAG services via the lazily
28# imported ``rag_service_factory`` (see _reconcile_unindexed_documents), so the
29# concrete ``LibraryRAGService`` is no longer imported at module scope.
30from ..database.library_init import get_default_library_id
31from ..database.models.library import Document, DocumentCollection
32from ..constants import DEFAULT_SEARCH_TOOL
35SCHEDULER_AVAILABLE = True # Always available since it's a required dependency
37# Per-tick cap for the opt-in library-collection index sweep. Bounds the work
38# done in a single scheduler thread tick so a large backlog of unindexed
39# documents self-heals gradually over successive ticks instead of blocking the
40# worker thread (the sweep is self-rate-limited by APScheduler max_instances=1
41# plus this batch cap).
42_LIBRARY_SWEEP_BATCH = 50
45class SchedulerCredentialStore(CredentialStoreBase):
46 """Credential store for the news scheduler.
48 Stores user passwords with TTL expiration so that background scheduler
49 jobs can access encrypted per-user databases.
50 """
52 def __init__(self, ttl_hours: int = 48):
53 super().__init__(ttl_hours * 3600)
55 def store(self, username: str, password: str) -> None:
56 """Store password for a user."""
57 self._store_credentials(
58 username, {"username": username, "password": password}
59 )
61 def retrieve(self, username: str) -> str | None:
62 """Retrieve password for a user. Returns None if expired/missing."""
63 result = self._retrieve_credentials(username, remove=False)
64 return result[1] if result else None
66 def clear(self, username: str) -> None:
67 """Clear stored password for a user."""
68 self.clear_entry(username)
71@dataclass(frozen=True)
72class DocumentSchedulerSettings:
73 """
74 Immutable settings snapshot for document scheduler.
76 Thread-safe: This is a frozen dataclass that can be safely passed
77 to and used from background threads.
78 """
80 enabled: bool = True
81 interval_seconds: int = 1800
82 download_pdfs: bool = False
83 extract_text: bool = True
84 generate_rag: bool = False
85 sweep_library_collections: bool = False
86 last_run: str = ""
88 @classmethod
89 def defaults(cls) -> "DocumentSchedulerSettings":
90 """Return default settings."""
91 return cls()
94class BackgroundJobScheduler:
95 """
96 Singleton scheduler that manages news subscriptions for active users.
98 This scheduler:
99 - Monitors user activity through database access
100 - Temporarily stores user credentials in memory
101 - Automatically schedules subscription checks
102 - Cleans up inactive users after configurable period
103 """
105 _instance = None
106 _lock = threading.Lock()
108 def __new__(cls):
109 """Ensure singleton instance."""
110 if cls._instance is None:
111 with cls._lock:
112 if cls._instance is None: 112 ↛ 114line 112 didn't jump to line 114
113 cls._instance = super().__new__(cls)
114 return cls._instance
116 def __init__(self):
117 """Initialize the scheduler (only runs once due to singleton)."""
118 # Skip if already initialized
119 if hasattr(self, "_initialized"):
120 return
122 # User session tracking
123 self.user_sessions = {} # user_id -> {last_activity, scheduled_jobs}
124 self.lock = threading.Lock()
126 # Credential store with TTL-based expiration
127 self._credential_store = SchedulerCredentialStore(ttl_hours=48)
129 # Scheduler instance
130 self.scheduler = BackgroundScheduler()
132 # Configuration (will be loaded from settings)
133 self.config = self._load_default_config()
135 # State
136 self.is_running = False
138 # Settings cache: username -> DocumentSchedulerSettings
139 # TTL of 300 seconds (5 minutes) reduces database queries
140 self._settings_cache: TTLCache = TTLCache(maxsize=100, ttl=300)
141 self._settings_cache_lock = threading.Lock()
143 self._initialized = True
144 logger.info("News scheduler initialized")
146 def _load_default_config(self) -> Dict[str, Any]:
147 """Load default configuration (will be overridden by settings manager)."""
148 return {
149 "enabled": True,
150 "retention_hours": 48,
151 "cleanup_interval_hours": 1,
152 "max_jitter_seconds": 300,
153 "max_concurrent_jobs": 10,
154 "subscription_batch_size": 5,
155 "activity_check_interval_minutes": 5,
156 }
158 def initialize_with_settings(self, settings_manager):
159 """Initialize configuration from settings manager."""
160 try:
161 # Load all scheduler settings
162 self.settings_manager = settings_manager
163 self.config = {
164 "enabled": self._get_setting("news.scheduler.enabled", True),
165 "retention_hours": self._get_setting(
166 "news.scheduler.retention_hours", 48
167 ),
168 "cleanup_interval_hours": self._get_setting(
169 "news.scheduler.cleanup_interval_hours", 1
170 ),
171 "max_jitter_seconds": self._get_setting(
172 "news.scheduler.max_jitter_seconds", 300
173 ),
174 "max_concurrent_jobs": self._get_setting(
175 "news.scheduler.max_concurrent_jobs", 10
176 ),
177 "subscription_batch_size": self._get_setting(
178 "news.scheduler.batch_size", 5
179 ),
180 "activity_check_interval_minutes": self._get_setting(
181 "news.scheduler.activity_check_interval", 5
182 ),
183 }
184 log_settings(self.config, "Scheduler configuration loaded")
185 except Exception:
186 logger.exception("Error loading scheduler settings")
187 # Keep default config
189 def _get_setting(self, key: str, default: Any) -> Any:
190 """Get setting with fallback to default."""
191 if hasattr(self, "settings_manager") and self.settings_manager:
192 return self.settings_manager.get_setting(key, default=default)
193 return default
195 def _wrap_job(
196 self, func: Callable, *, username: str | None = None
197 ) -> Callable:
198 """Wrap a scheduler job so it runs inside a ``request_user`` context.
200 APScheduler runs jobs in a thread pool. Threads do NOT inherit
201 contextvars from the scheduling thread by default, so without
202 explicit propagation, ``get_current_username()`` returns ``None``
203 inside scheduled work — breaking metrics/log attribution and any
204 service-layer code that consults the contextvar.
206 For user-bound jobs (subscription checks, document processing),
207 callers pass ``username=<value>`` so the wrapper pushes a
208 ``request_user`` context that survives for the duration of the
209 job. For system jobs (cleanup, config reload) ``username`` stays
210 ``None`` and no contextvar is pushed.
211 """
212 from contextlib import nullcontext
214 from ..utilities.request_context import request_user
216 @wraps(func)
217 def wrapper(*args, **kwargs):
218 user_ctx = request_user(username) if username else nullcontext()
219 with user_ctx:
220 return func(*args, **kwargs)
222 return wrapper
224 def _get_document_scheduler_settings(
225 self, username: str, force_refresh: bool = False
226 ) -> DocumentSchedulerSettings:
227 """
228 Get document scheduler settings for a user with TTL caching.
230 This is the single source of truth for document scheduler settings.
231 Settings are cached for 5 minutes by default to reduce database queries.
233 Args:
234 username: User to get settings for
235 force_refresh: If True, bypass cache and fetch fresh settings
237 Returns:
238 DocumentSchedulerSettings dataclass (frozen/immutable for thread-safety)
239 """
240 # Fast path: check cache without modifying it
241 if not force_refresh:
242 with self._settings_cache_lock:
243 cached = self._settings_cache.get(username)
244 if cached is not None:
245 logger.debug(f"[SETTINGS_CACHE] Cache hit for {username}")
246 cached_settings: DocumentSchedulerSettings = cached
247 return cached_settings
249 # Cache miss - need to fetch from database
250 logger.debug(
251 f"[SETTINGS_CACHE] Cache miss for {username}, fetching from DB"
252 )
254 # Get password from session
255 session_info = self.user_sessions.get(username)
256 if not session_info:
257 logger.warning(
258 f"[SETTINGS_CACHE] No session info for {username}, using defaults"
259 )
260 return DocumentSchedulerSettings.defaults()
262 password = self._credential_store.retrieve(username)
263 if not password: 263 ↛ 264line 263 didn't jump to line 264 because the condition on line 263 was never true
264 logger.warning(
265 f"[SETTINGS_CACHE] Credentials expired for {username}, using defaults"
266 )
267 return DocumentSchedulerSettings.defaults()
269 # Fetch settings from database (outside lock to avoid blocking)
270 try:
271 from ..database.session_context import get_user_db_session
272 from ..settings.manager import SettingsManager
274 with get_user_db_session(username, password) as db:
275 sm = SettingsManager(db)
277 settings = DocumentSchedulerSettings(
278 enabled=sm.get_setting("document_scheduler.enabled", True),
279 interval_seconds=sm.get_setting(
280 "document_scheduler.interval_seconds", 1800
281 ),
282 download_pdfs=sm.get_setting(
283 "document_scheduler.download_pdfs", False
284 ),
285 extract_text=sm.get_setting(
286 "document_scheduler.extract_text", True
287 ),
288 generate_rag=sm.get_setting(
289 "document_scheduler.generate_rag", False
290 ),
291 sweep_library_collections=sm.get_setting(
292 "document_scheduler.sweep_library_collections", False
293 ),
294 last_run=sm.get_setting("document_scheduler.last_run", ""),
295 )
297 # Store in cache
298 with self._settings_cache_lock:
299 self._settings_cache[username] = settings
300 logger.debug(f"[SETTINGS_CACHE] Cached settings for {username}")
302 return settings
304 except Exception as e:
305 # ``password`` (retrieved above) is in the frame locals of
306 # this handler and of ``get_user_db_session``; a traceback
307 # rendered with diagnose=True would leak it. Drop the
308 # traceback chain and redact str(e).
309 safe_msg = redact_secrets(str(e), password)
310 logger.warning(
311 f"[SETTINGS_CACHE] Error fetching settings for {username}: {safe_msg}"
312 )
313 return DocumentSchedulerSettings.defaults()
315 def invalidate_user_settings_cache(self, username: str) -> bool:
316 """
317 Invalidate cached settings for a specific user.
319 Call this when user settings change or user logs out.
321 Args:
322 username: User whose cache to invalidate
324 Returns:
325 True if cache entry was removed, False if not found
326 """
327 with self._settings_cache_lock:
328 if username in self._settings_cache:
329 del self._settings_cache[username]
330 logger.debug(
331 f"[SETTINGS_CACHE] Invalidated cache for {username}"
332 )
333 return True
334 return False
336 def invalidate_all_settings_cache(self) -> int:
337 """
338 Invalidate all cached settings.
340 Call this when doing bulk settings updates or during config reload.
342 Returns:
343 Number of cache entries cleared
344 """
345 with self._settings_cache_lock:
346 count = len(self._settings_cache)
347 self._settings_cache.clear()
348 logger.info(
349 f"[SETTINGS_CACHE] Cleared all settings cache ({count} entries)"
350 )
351 return count
353 def start(self):
354 """Start the scheduler."""
355 if not self.config.get("enabled", True):
356 logger.info("News scheduler is disabled in settings")
357 return
359 if self.is_running:
360 logger.warning("Scheduler is already running")
361 return
363 # Schedule cleanup job
364 self.scheduler.add_job(
365 self._wrap_job(self._run_cleanup_with_tracking),
366 "interval",
367 hours=self.config["cleanup_interval_hours"],
368 id="cleanup_inactive_users",
369 name="Cleanup Inactive User Sessions",
370 jitter=60, # Add some jitter to cleanup
371 )
373 # Schedule configuration reload
374 self.scheduler.add_job(
375 self._wrap_job(self._reload_config),
376 "interval",
377 minutes=30,
378 id="reload_config",
379 name="Reload Configuration",
380 )
382 # Start the scheduler
383 self.scheduler.start()
384 self.is_running = True
386 # Schedule initial cleanup after a delay
387 self.scheduler.add_job(
388 self._wrap_job(self._run_cleanup_with_tracking),
389 "date",
390 run_date=datetime.now(UTC) + timedelta(seconds=30),
391 id="initial_cleanup",
392 )
394 logger.info("News scheduler started")
396 def stop(self):
397 """Stop the scheduler."""
398 if self.is_running:
399 self.scheduler.shutdown(wait=True)
400 self.is_running = False
402 # Clear all user sessions and credentials
403 with self.lock:
404 for username in self.user_sessions:
405 self._credential_store.clear(username)
406 self.user_sessions.clear()
408 logger.info("News scheduler stopped")
410 def update_user_info(self, username: str, password: str):
411 """
412 Update user info in scheduler. Called on every database interaction.
414 Args:
415 username: User's username
416 password: User's password
417 """
418 logger.info(
419 f"[SCHEDULER] update_user_info called for {username}, is_running={self.is_running}, active_users={len(self.user_sessions)}"
420 )
421 logger.debug(
422 f"[SCHEDULER] Current active users: {list(self.user_sessions.keys())}"
423 )
425 if not self.is_running:
426 logger.warning(
427 f"[SCHEDULER] Scheduler not running, cannot update user {username}"
428 )
429 return
431 with self.lock:
432 # Store password in credential store (inside lock to prevent
433 # race where concurrent calls leave mismatched credentials)
434 self._credential_store.store(username, password)
436 now = datetime.now(UTC)
438 if username not in self.user_sessions:
439 # New user - create session info
440 logger.info(f"[SCHEDULER] New user in scheduler: {username}")
441 self.user_sessions[username] = {
442 "last_activity": now,
443 "scheduled_jobs": set(),
444 }
445 logger.debug(
446 f"[SCHEDULER] Created session for {username}, scheduling subscriptions"
447 )
448 # Schedule their subscriptions
449 self._schedule_user_subscriptions(username)
450 else:
451 # Existing user - update info
452 logger.info(
453 f"[SCHEDULER] Updating existing user {username} activity, will reschedule"
454 )
455 old_activity = self.user_sessions[username]["last_activity"]
456 activity_delta = now - old_activity
457 logger.debug(
458 f"[SCHEDULER] User {username} last activity: {old_activity}, delta: {activity_delta}"
459 )
461 self.user_sessions[username]["last_activity"] = now
462 logger.debug(
463 f"[SCHEDULER] Updated {username} session info, scheduling subscriptions"
464 )
465 # Reschedule their subscriptions in case they changed
466 self._schedule_user_subscriptions(username)
468 def unregister_user(self, username: str):
469 """
470 Unregister a user and clean up their scheduled jobs.
471 Called when user logs out.
472 """
473 with self.lock:
474 if username in self.user_sessions:
475 logger.info(f"Unregistering user {username}")
477 # Remove all scheduled jobs for this user
478 session_info = self.user_sessions[username]
479 for job_id in session_info["scheduled_jobs"].copy():
480 try:
481 self.scheduler.remove_job(job_id)
482 except JobLookupError:
483 pass
485 # Remove user session and clear credentials atomically
486 del self.user_sessions[username]
487 self._credential_store.clear(username)
489 # Invalidate settings cache for this user (outside lock)
490 self.invalidate_user_settings_cache(username)
491 logger.info(f"User {username} unregistered successfully")
493 def reschedule_document_jobs(self, username: str) -> bool:
494 """(Re)schedule the document-processing + reconciler jobs for an
495 ACTIVE user against their current settings.
497 Call this after a ``document_scheduler.*`` setting changes so the
498 change takes effect on the next interval tick instead of only after
499 the user logs out and back in. Without it, ``_schedule_reconciler``
500 runs only via the login path (``update_user_info`` ->
501 ``_schedule_user_subscriptions`` -> ``_schedule_document_processing``):
502 the runtime gate inside ``_reconcile_unindexed_documents`` neutralises
503 a stale job after a *disable*, but an *enable* (including toggling the
504 legacy ``generate_rag`` arm, which on older builds took effect on the
505 next tick) would otherwise never create the ``{username}_library_sweep``
506 job until the next login.
508 Relies on the cache having already been invalidated by the caller
509 (``invalidate_settings_caches``) so ``_schedule_document_processing``
510 re-reads fresh settings. No password argument is needed: the job is
511 rebuilt from the credentials the scheduler already holds for the active
512 session. Returns ``True`` if a reschedule was performed, ``False`` for
513 a user the scheduler isn't tracking (their jobs are built from current
514 settings on their next login) or when the scheduler isn't running.
515 """
516 if not self.is_running:
517 return False
518 with self.lock:
519 if username not in self.user_sessions:
520 logger.debug(
521 f"[DOC_SCHEDULER] reschedule_document_jobs: {username} "
522 "not an active scheduler session; skipping"
523 )
524 return False
525 self.user_sessions[username]["last_activity"] = datetime.now(UTC)
526 # Re-reads fresh settings (cache invalidated by the caller) and
527 # re-adds the document-processing job + reconciler per current
528 # settings; both use replace_existing=True so this is idempotent.
529 self._schedule_document_processing(username)
530 return True
532 def reschedule_zotero_jobs(self, username: str) -> bool:
533 """(Re)schedule the Zotero auto-sync job for an ACTIVE user against
534 their current settings.
536 Call this after a ``zotero.*`` setting changes (e.g. toggling
537 ``auto_sync_enabled`` or changing ``sync_interval_minutes``) so the
538 change takes effect on the next tick instead of only after the user logs
539 out and back in — otherwise ``_schedule_zotero_sync`` runs only via the
540 login path (``update_user_info`` -> ``_schedule_user_subscriptions``).
541 ``_schedule_zotero_sync`` already removes any existing job and no-ops
542 when auto-sync is disabled, so this both creates and tears down as
543 needed and is idempotent (``replace_existing=True``).
545 Relies on the caller having invalidated settings caches first so
546 ``get_config()`` re-reads fresh settings. Returns ``True`` if a
547 reschedule was performed, ``False`` for a user the scheduler isn't
548 tracking (built from current settings on their next login) or when the
549 scheduler isn't running.
550 """
551 if not self.is_running:
552 return False
553 with self.lock:
554 if username not in self.user_sessions:
555 logger.debug(
556 f"[ZOTERO_SCHEDULER] reschedule_zotero_jobs: {username} "
557 "not an active scheduler session; skipping"
558 )
559 return False
560 self.user_sessions[username]["last_activity"] = datetime.now(UTC)
561 self._schedule_zotero_sync(username)
562 return True
564 def _schedule_user_subscriptions(self, username: str):
565 """Schedule all active subscriptions for a user."""
566 logger.info(f"_schedule_user_subscriptions called for {username}")
567 # Pre-declared so the leak-redaction in the except handler is safe
568 # if the exception fires before ``password`` is assigned below.
569 password = None
570 try:
571 session_info = self.user_sessions.get(username)
572 if not session_info:
573 logger.warning(f"No session info found for {username}")
574 return
576 password = self._credential_store.retrieve(username)
577 if not password: 577 ↛ 578line 577 didn't jump to line 578 because the condition on line 577 was never true
578 logger.warning(
579 f"Credentials expired for {username}, skipping subscription scheduling"
580 )
581 return
582 logger.debug(f"Got password for {username}: present")
584 # Get user's subscriptions from their encrypted database
585 from ..database.session_context import get_user_db_session
586 from ..database.models.news import NewsSubscription
588 with get_user_db_session(username, password) as db:
589 subscriptions = (
590 db.query(NewsSubscription)
591 .filter(NewsSubscription.active_filter())
592 .all()
593 )
594 logger.debug(
595 f"Query executed, found {len(subscriptions)} results"
596 )
598 # Log details of each subscription
599 for sub in subscriptions:
600 logger.debug(
601 f"Subscription {sub.id}: name='{sub.name}', status='{sub.status}', refresh_interval={sub.refresh_interval_minutes} minutes"
602 )
604 logger.info(
605 f"Found {len(subscriptions)} active subscriptions for {username}"
606 )
608 # Clear old jobs for this user
609 for job_id in session_info["scheduled_jobs"].copy():
610 try:
611 self.scheduler.remove_job(job_id)
612 session_info["scheduled_jobs"].remove(job_id)
613 except JobLookupError:
614 pass
616 # Schedule each subscription with jitter
617 for sub in subscriptions:
618 job_id = f"{username}_{sub.id}"
620 # Calculate jitter
621 # Security: random jitter to distribute subscription timing, not security-sensitive
622 max_jitter = int(self.config.get("max_jitter_seconds", 300))
623 jitter = random.randint(0, max_jitter)
625 # Determine trigger based on frequency
626 refresh_minutes = sub.refresh_interval_minutes
628 if refresh_minutes <= 60: # 60 minutes or less
629 # For hourly or more frequent, use interval trigger
630 trigger = "interval"
631 trigger_args = {
632 "minutes": refresh_minutes,
633 "jitter": jitter,
634 "start_date": datetime.now(UTC), # Start immediately
635 }
636 else:
637 # For less frequent, calculate next run time
638 now = datetime.now(UTC)
639 if sub.next_refresh:
640 # Ensure timezone-aware for comparison with now (UTC)
641 next_refresh_aware = sub.next_refresh
642 if next_refresh_aware.tzinfo is None: 642 ↛ 650line 642 didn't jump to line 650 because the condition on line 642 was always true
643 logger.warning(
644 f"Subscription {sub.id} has naive (non-tz-aware) "
645 f"next_refresh datetime, assuming UTC"
646 )
647 next_refresh_aware = next_refresh_aware.replace(
648 tzinfo=UTC
649 )
650 if next_refresh_aware <= now: 650 ↛ 652line 650 didn't jump to line 652 because the condition on line 650 was never true
651 # Subscription is overdue - run it immediately with small jitter
652 logger.info(
653 f"Subscription {sub.id} is overdue, scheduling immediate run"
654 )
655 next_run = now + timedelta(seconds=jitter)
656 else:
657 next_run = next_refresh_aware
658 else:
659 next_run = now + timedelta(
660 minutes=refresh_minutes, seconds=jitter
661 )
663 trigger = "date"
664 trigger_args = {"run_date": next_run}
666 # Add the job
667 self.scheduler.add_job(
668 func=self._wrap_job(
669 self._check_subscription, username=username
670 ),
671 args=[username, sub.id],
672 trigger=trigger,
673 id=job_id,
674 name=f"Check {sub.name or sub.query_or_topic[:30]}",
675 replace_existing=True,
676 **trigger_args,
677 )
679 session_info["scheduled_jobs"].add(job_id)
680 logger.info(f"Scheduled job {job_id} with {trigger} trigger")
682 except Exception as e:
683 # ``password`` was retrieved from the credential store
684 # above (line ~483) and passed into ``get_user_db_session``.
685 # An exception from the DB session (e.g. SQLCipher
686 # ``OperationalError``) can carry frame locals that include
687 # the plaintext SQLCipher master password — which is
688 # unrecoverable (TRUST.md §5). Drop the traceback chain and
689 # redact str(e).
690 safe_msg = redact_secrets(str(e), password)
691 logger.warning(
692 f"Error scheduling subscriptions for {username}: {safe_msg}"
693 )
695 # Add document processing for this user
696 self._schedule_document_processing(username)
698 # Add Zotero auto-sync for this user (no-op unless enabled)
699 self._schedule_zotero_sync(username)
701 def _schedule_document_processing(self, username: str):
702 """Schedule document processing for a user."""
703 logger.info(
704 f"[DOC_SCHEDULER] Scheduling document processing for {username}"
705 )
706 logger.debug(
707 f"[DOC_SCHEDULER] Current user sessions: {list(self.user_sessions.keys())}"
708 )
710 try:
711 session_info = self.user_sessions.get(username)
712 if not session_info:
713 logger.warning(
714 f"[DOC_SCHEDULER] No session info found for {username}"
715 )
716 logger.debug(
717 f"[DOC_SCHEDULER] Available sessions: {list(self.user_sessions.keys())}"
718 )
719 return
721 logger.debug(
722 f"[DOC_SCHEDULER] Retrieved session for {username}, scheduler running: {self.is_running}"
723 )
725 # Get user's document scheduler settings (cached)
726 settings = self._get_document_scheduler_settings(username)
728 if not settings.enabled:
729 logger.info(
730 f"[DOC_SCHEDULER] Document scheduler disabled for user {username}"
731 )
732 return
734 logger.info(
735 f"[DOC_SCHEDULER] User {username} document settings: enabled={settings.enabled}, "
736 f"interval={settings.interval_seconds}s, pdfs={settings.download_pdfs}, "
737 f"text={settings.extract_text}, "
738 f"index={settings.generate_rag or settings.sweep_library_collections}"
739 )
741 # Schedule document processing job
742 job_id = f"{username}_document_processing"
743 logger.debug(f"[DOC_SCHEDULER] Preparing to schedule job {job_id}")
745 # Remove existing document job if any
746 try:
747 self.scheduler.remove_job(job_id)
748 session_info["scheduled_jobs"].discard(job_id)
749 logger.debug(f"[DOC_SCHEDULER] Removed existing job {job_id}")
750 except JobLookupError:
751 logger.debug(
752 f"[DOC_SCHEDULER] No existing job {job_id} to remove"
753 )
754 pass # Job doesn't exist, that's fine
756 # Add new document processing job
757 logger.debug(
758 f"[DOC_SCHEDULER] Adding new document processing job with interval {settings.interval_seconds}s"
759 )
760 self.scheduler.add_job(
761 func=self._wrap_job(
762 self._process_user_documents, username=username
763 ),
764 args=[username],
765 trigger="interval",
766 seconds=settings.interval_seconds,
767 id=job_id,
768 name=f"Process Documents for {username}",
769 jitter=30, # Add small jitter to prevent multiple users from processing simultaneously
770 max_instances=1, # Prevent overlapping document processing for same user
771 replace_existing=True,
772 )
774 session_info["scheduled_jobs"].add(job_id)
775 logger.info(
776 f"[DOC_SCHEDULER] Scheduled document processing job {job_id} for {username} with {settings.interval_seconds}s interval"
777 )
778 logger.debug(
779 f"[DOC_SCHEDULER] User {username} now has {len(session_info['scheduled_jobs'])} scheduled jobs: {list(session_info['scheduled_jobs'])}"
780 )
782 # Verify job was added
783 job = self.scheduler.get_job(job_id)
784 if job:
785 logger.info(
786 f"[DOC_SCHEDULER] Successfully verified job {job_id} exists, next run: {job.next_run_time}"
787 )
788 else:
789 logger.error(
790 f"[DOC_SCHEDULER] Failed to verify job {job_id} exists!"
791 )
793 # Schedule (or tear down) the unindexed-document reconciler,
794 # mirroring the document-processing job's lifecycle.
795 self._schedule_reconciler(username, settings, session_info)
797 except Exception as e:
798 # No ``password`` local here, but the caller frame
799 # (``_schedule_user_subscriptions``) holds the SQLCipher
800 # master password — loguru ``diagnose=True`` walks the
801 # frame stack and would render that caller-frame local.
802 # Drop the traceback by using ``logger.warning`` without
803 # ``exc_info``. ``redact_secrets`` with ``None`` is a no-op
804 # here, but kept for the check-sensitive-logging pre-commit
805 # hook + as a guide-post pairing for future refactors that
806 # might bring a password into scope.
807 safe_msg = redact_secrets(str(e), None)
808 logger.warning(
809 f"Error scheduling document processing for {username}: {safe_msg}"
810 )
812 def _schedule_reconciler(
813 self,
814 username: str,
815 settings: DocumentSchedulerSettings,
816 session_info: Dict[str, Any],
817 ) -> None:
818 """Add or remove the unindexed-document reconciler job.
820 Mirrors the document-processing job lifecycle in
821 ``_schedule_document_processing``: the job is (re)created only when
822 EITHER ``sweep_library_collections`` OR ``generate_rag`` is enabled, and
823 removed (and dropped from the session's tracked-jobs set) when both are
824 off — so toggling the settings off and rescheduling tears the job down
825 cleanly. The reconciler indexes every unindexed document (uploaded
826 library docs AND research downloads), so both settings gate it: the
827 ``generate_rag`` OR-arm preserves the legacy "index research downloads"
828 behaviour that used to live inline in ``_process_user_documents``.
829 """
830 job_id = f"{username}_library_sweep"
832 # Always remove any existing instance first so a disabled setting
833 # tears the job down and a changed interval is re-applied.
834 try:
835 self.scheduler.remove_job(job_id)
836 session_info["scheduled_jobs"].discard(job_id)
837 logger.debug(f"[RECONCILER] Removed existing job {job_id}")
838 except JobLookupError:
839 pass # Job doesn't exist, that's fine
841 if not (settings.sweep_library_collections or settings.generate_rag):
842 logger.debug(
843 f"[RECONCILER] Indexing disabled for {username}; not scheduling"
844 )
845 return
847 self.scheduler.add_job(
848 func=self._wrap_job(
849 self._reconcile_unindexed_documents, username=username
850 ),
851 args=[username],
852 trigger="interval",
853 seconds=settings.interval_seconds,
854 id=job_id,
855 name=f"Unindexed Document Reconciler for {username}",
856 jitter=60,
857 max_instances=1, # Self-rate-limit: no overlapping runs
858 replace_existing=True,
859 )
860 session_info["scheduled_jobs"].add(job_id)
861 logger.info(
862 f"[RECONCILER] Scheduled unindexed-document reconciler job {job_id} "
863 f"for {username} with {settings.interval_seconds}s interval"
864 )
866 def _arm_egress_backstop(self, settings_manager, username: str) -> bool:
867 """Set the audit-hook egress context from the user's saved settings so
868 scheduled document downloads run under the same secondary net as an
869 interactive research run. Returns False when policy setup itself fails,
870 because scheduled document work must not run without its policy
871 snapshot. Other backstop failures remain best-effort; the
872 DownloadService PEP remains the primary gate. Cleared by the caller's
873 @thread_cleanup on exit.
874 """
875 try:
876 from ..security.egress.audit_hook import set_active_context
877 from ..security.egress.policy import (
878 context_from_snapshot,
879 resolve_run_primary_engine,
880 )
882 snapshot = settings_manager.get_settings_snapshot(strict=True)
883 if not isinstance(snapshot, dict):
884 logger.bind(policy_audit=True).warning(
885 "doc scheduler: egress policy setup failed; skipping scheduled work"
886 )
887 return False
888 # Derive the primary engine from the SAME strict snapshot above
889 # rather than a second, non-strict ``get_setting("search.tool",
890 # DEFAULT_SEARCH_TOOL)`` call. A non-strict read could fall back to
891 # the public DEFAULT_SEARCH_TOOL and silently widen the run's
892 # adaptive scope. ``resolve_run_primary_engine`` is the single
893 # source of truth used by the run-scoped PEPs; with no ``default``
894 # it raises ValueError when ``search.tool`` is missing/blank, which
895 # the outer except catches → fail-closed (no scheduled work).
896 primary = resolve_run_primary_engine(snapshot)
897 ctx = context_from_snapshot(snapshot, primary, username=username)
898 except Exception:
899 logger.bind(policy_audit=True).warning(
900 "doc scheduler: egress policy setup failed; skipping scheduled work"
901 )
902 return False
904 try:
905 set_active_context(ctx)
906 except Exception:
907 logger.bind(policy_audit=True).debug(
908 "doc scheduler: egress backstop not armed", exc_info=True
909 )
910 return True
912 @thread_cleanup
913 def _process_user_documents(self, username: str):
914 """Process documents for a user."""
915 logger.info(f"[DOC_SCHEDULER] Processing documents for user {username}")
916 start_time = datetime.now(UTC)
918 # Pre-declared so the except handlers can pass it to redact_secrets
919 # even if the retrieve() call below itself raises.
920 password = None
921 try:
922 session_info = self.user_sessions.get(username)
923 if not session_info:
924 logger.warning(
925 f"[DOC_SCHEDULER] No session info found for user {username}"
926 )
927 return
929 password = self._credential_store.retrieve(username)
930 if not password:
931 logger.warning(
932 f"[DOC_SCHEDULER] Credentials expired for user {username}"
933 )
934 return
935 logger.debug(
936 f"[DOC_SCHEDULER] Starting document processing for {username}"
937 )
939 # Get user's document scheduler settings (cached)
940 settings = self._get_document_scheduler_settings(username)
942 logger.info(
943 f"[DOC_SCHEDULER] Processing settings for {username}: "
944 f"pdfs={settings.download_pdfs}, text={settings.extract_text}"
945 )
947 # RAG indexing has moved to ``_reconcile_unindexed_documents`` (its
948 # own scheduled job), so ``generate_rag`` no longer drives any work
949 # in this download/extract pass. Only the file-producing passes gate
950 # whether this method runs.
951 if not any(
952 [
953 settings.download_pdfs,
954 settings.extract_text,
955 ]
956 ):
957 logger.info(
958 f"[DOC_SCHEDULER] No download/extract options enabled for user {username}"
959 )
960 return
962 # Parse last_run from cached settings
963 last_run = (
964 datetime.fromisoformat(settings.last_run)
965 if settings.last_run
966 else None
967 )
969 logger.info(f"[DOC_SCHEDULER] Last run for {username}: {last_run}")
971 # Need database session for queries and updates
972 from ..database.session_context import get_user_db_session
973 from ..database.models.research import ResearchHistory
974 from ..settings.manager import SettingsManager
976 with get_user_db_session(username, password) as db:
977 settings_manager = SettingsManager(db)
979 # Arm the PEP-578 audit-hook backstop for this scheduled run.
980 # The APScheduler worker thread carries no egress context, so
981 # the secondary net would be inactive while DownloadService
982 # fetches documents below. DownloadService's evaluate_url PEP
983 # still gates each fetch (primary); this restores defense-in-
984 # depth parity with an interactive run. @thread_cleanup clears
985 # the context when this method returns.
986 if not self._arm_egress_backstop(settings_manager, username):
987 return
989 # Query for completed research since last run
990 logger.debug(
991 f"[DOC_SCHEDULER] Querying for completed research since {last_run}"
992 )
993 query = db.query(ResearchHistory).filter(
994 ResearchHistory.status == ResearchStatus.COMPLETED,
995 ResearchHistory.completed_at.is_not(
996 None
997 ), # Ensure completed_at is not null
998 )
1000 if last_run:
1001 query = query.filter(
1002 ResearchHistory.completed_at > last_run
1003 )
1005 # Limit to recent research to prevent overwhelming
1006 query = query.order_by(
1007 ResearchHistory.completed_at.desc()
1008 ).limit(20)
1010 research_sessions = query.all()
1011 logger.debug(
1012 f"[DOC_SCHEDULER] Query executed, found {len(research_sessions)} sessions"
1013 )
1015 if not research_sessions:
1016 logger.info(
1017 f"[DOC_SCHEDULER] No new completed research sessions found for user {username}"
1018 )
1019 return
1021 logger.info(
1022 f"[DOC_SCHEDULER] Found {len(research_sessions)} research sessions to process for {username}"
1023 )
1025 # Log details of each research session
1026 for i, research in enumerate(
1027 research_sessions[:5]
1028 ): # Log first 5 details
1029 title_safe = (
1030 (research.title[:50] + "...")
1031 if research.title
1032 else "No title"
1033 )
1034 completed_safe = (
1035 research.completed_at
1036 if research.completed_at
1037 else "No completion time"
1038 )
1039 logger.debug(
1040 f"[DOC_SCHEDULER] Session {i + 1}: id={research.id}, title={title_safe}, completed={completed_safe}"
1041 )
1043 # Handle completed_at which might be a string or datetime
1044 completed_at_obj = None
1045 if research.completed_at:
1046 if isinstance(research.completed_at, str):
1047 try:
1048 completed_at_obj = datetime.fromisoformat(
1049 research.completed_at.replace("Z", "+00:00")
1050 )
1051 except (ValueError, TypeError, AttributeError):
1052 completed_at_obj = None
1053 else:
1054 completed_at_obj = research.completed_at
1056 logger.debug(
1057 f"[DOC_SCHEDULER] - completed_at type: {type(research.completed_at)}"
1058 )
1059 logger.debug(
1060 f"[DOC_SCHEDULER] - completed_at timezone: {completed_at_obj.tzinfo if completed_at_obj else 'None'}"
1061 )
1062 logger.debug(f"[DOC_SCHEDULER] - last_run: {last_run}")
1063 logger.debug(
1064 f"[DOC_SCHEDULER] - completed_at > last_run: {completed_at_obj > last_run if last_run and completed_at_obj else 'N/A'}"
1065 )
1067 # Capture a settings snapshot for this user/run so the
1068 # DownloadService below can build an EgressContext and
1069 # gate each per-resource URL. Without this the scheduler
1070 # would bypass policy entirely. Reuses the outer `db`
1071 # session (line 743) — get_settings_manager() in a
1072 # background thread must be passed a db_session
1073 # explicitly per the pre-commit thread-safety check.
1074 try:
1075 user_settings_snapshot = (
1076 settings_manager.get_settings_snapshot(strict=True)
1077 )
1078 except Exception:
1079 logger.bind(policy_audit=True).warning(
1080 "doc scheduler: egress policy setup failed; skipping scheduled work"
1081 )
1082 return
1084 if not isinstance(user_settings_snapshot, dict):
1085 logger.bind(policy_audit=True).warning(
1086 "doc scheduler: egress policy setup failed; skipping scheduled work"
1087 )
1088 return
1090 processed_count = 0
1091 for research in research_sessions:
1092 try:
1093 logger.info(
1094 f"[DOC_SCHEDULER] Processing research {research.id} for user {username}"
1095 )
1097 # Set search context so rate limiting works in both
1098 # download_pdfs and extract_text paths
1099 from ..utilities.thread_context import (
1100 set_search_context,
1101 )
1103 set_search_context(
1104 {
1105 "research_id": str(research.id),
1106 "username": username,
1107 "user_password": password,
1108 "research_phase": "document_scheduler",
1109 }
1110 )
1112 # Call actual processing APIs
1113 if settings.download_pdfs:
1114 logger.info(
1115 f"[DOC_SCHEDULER] Downloading PDFs for research {research.id}"
1116 )
1117 try:
1118 # Use the DownloadService to queue PDF downloads
1119 from ..research_library.services.download_service import (
1120 DownloadService,
1121 )
1123 with DownloadService(
1124 username,
1125 password,
1126 settings_snapshot=user_settings_snapshot,
1127 ) as download_service:
1128 queued_count = download_service.queue_research_downloads(
1129 research.id
1130 )
1131 logger.info(
1132 f"[DOC_SCHEDULER] Queued {queued_count} PDF downloads for research {research.id}"
1133 )
1134 except Exception as e:
1135 # Recover the shared thread-local session
1136 # before continuing — without rollback the
1137 # next phase (text extract / RAG) and the
1138 # post-loop last_run commit run on a
1139 # poisoned session (issue #3827).
1140 safe_rollback(db, "DOC_SCHEDULER PDF download")
1141 # ``password`` is in scope and was passed
1142 # into ``DownloadService``. Drop traceback
1143 # + redact str(e) to avoid leaking the
1144 # SQLCipher master password under
1145 # ``diagnose=True``.
1146 safe_msg = redact_secrets(str(e), password)
1147 logger.warning(
1148 f"[DOC_SCHEDULER] Failed to download PDFs for research {research.id}: {safe_msg}"
1149 )
1151 if settings.extract_text:
1152 logger.info(
1153 f"[DOC_SCHEDULER] Extracting text for research {research.id}"
1154 )
1155 try:
1156 # Use the DownloadService to extract text for all resources
1157 from ..research_library.services.download_service import (
1158 DownloadService,
1159 )
1160 from ..database.models.research import (
1161 ResearchResource,
1162 )
1164 from ..research_library.utils import (
1165 is_downloadable_url,
1166 )
1168 with DownloadService(
1169 username,
1170 password,
1171 settings_snapshot=user_settings_snapshot,
1172 ) as download_service:
1173 # Get all resources for this research (reuse existing db session)
1174 all_resources = (
1175 db.query(ResearchResource)
1176 .filter_by(research_id=research.id)
1177 .all()
1178 )
1179 # Filter: only process downloadable resources (academic/PDF)
1180 resources = [
1181 r
1182 for r in all_resources
1183 if is_downloadable_url(r.url)
1184 ]
1185 processed_count = 0
1186 for resource in resources:
1187 # We need to pass the password to the download service
1188 # The DownloadService creates its own database sessions, so we need to ensure password is available
1189 try:
1190 success, error = (
1191 download_service.download_as_text(
1192 resource.id
1193 )
1194 )
1195 if success:
1196 processed_count += 1
1197 logger.info(
1198 f"[DOC_SCHEDULER] Successfully extracted text for resource {resource.id}"
1199 )
1200 else:
1201 logger.warning(
1202 f"[DOC_SCHEDULER] Failed to extract text for resource {resource.id}: {error}"
1203 )
1204 except Exception as resource_error:
1205 # Roll back FIRST so the next
1206 # iteration's queries don't
1207 # cascade on a poisoned session
1208 # (issue #3827).
1209 safe_rollback(
1210 db,
1211 "DOC_SCHEDULER resource",
1212 )
1213 # ``password`` is in scope and
1214 # was passed into the enclosing
1215 # ``DownloadService``. Drop the
1216 # traceback chain + redact str(e)
1217 # to avoid leaking the SQLCipher
1218 # master password.
1219 safe_msg = redact_secrets(
1220 str(resource_error), password
1221 )
1222 logger.warning(
1223 f"[DOC_SCHEDULER] Error processing resource {resource.id}: {safe_msg}"
1224 )
1225 logger.info(
1226 f"[DOC_SCHEDULER] Text extraction completed for research {research.id}: {processed_count}/{len(resources)} resources processed"
1227 )
1228 except Exception as e:
1229 safe_rollback(
1230 db, "DOC_SCHEDULER text extraction"
1231 )
1232 # ``password`` is in scope from the outer
1233 # ``_process_user_documents`` retrieval —
1234 # same redact + warning pattern as the
1235 # inner handlers in this function.
1236 safe_msg = redact_secrets(str(e), password)
1237 logger.warning(
1238 f"[DOC_SCHEDULER] Failed to extract text for research {research.id}: {safe_msg}"
1239 )
1241 # NOTE: RAG indexing of research downloads used to live
1242 # here (the old ``if settings.generate_rag:`` block).
1243 # It has been retired — the unified
1244 # ``_reconcile_unindexed_documents`` reconciler now
1245 # indexes ALL unindexed documents (including research
1246 # downloads that have no DocumentCollection row yet) on
1247 # its own schedule, gated by ``generate_rag OR
1248 # sweep_library_collections``. The download_pdfs and
1249 # extract_text passes above remain here because they
1250 # produce the ``text_content`` the reconciler indexes.
1252 processed_count += 1
1253 logger.debug(
1254 f"[DOC_SCHEDULER] Successfully queued processing for research {research.id}"
1255 )
1257 except Exception as e:
1258 safe_rollback(db, "DOC_SCHEDULER research")
1259 # ``password`` is in scope from the outer
1260 # ``_process_user_documents`` retrieval. Drop the
1261 # traceback chain and redact str(e).
1262 safe_msg = redact_secrets(str(e), password)
1263 logger.warning(
1264 f"[DOC_SCHEDULER] Error processing research {research.id} for user {username}: {safe_msg}"
1265 )
1267 # Update last run time in user's settings.
1268 # Intentionally NOT wrapped in try/finally: if upstream setup
1269 # fails (DB open, SettingsManager init, initial query),
1270 # last_run should stay put so the next tick retries.
1271 # Advancing here would mask a persistent failure (corrupted
1272 # DB, wrong password). See closed PR #3288.
1273 current_time = datetime.now(UTC).isoformat()
1274 settings_manager.set_setting(
1275 "document_scheduler.last_run", current_time, commit=True
1276 )
1277 logger.debug(
1278 f"[DOC_SCHEDULER] Updated last run time for {username} to {current_time}"
1279 )
1281 end_time = datetime.now(UTC)
1282 duration = (end_time - start_time).total_seconds()
1283 logger.info(
1284 f"[DOC_SCHEDULER] Completed document processing for user {username}: {processed_count} sessions processed in {duration:.2f}s"
1285 )
1287 except Exception as e:
1288 # ``password`` is pre-declared as ``None`` at the top of the
1289 # function, so it is always bound here even if the retrieve()
1290 # call itself raised. ``redact_secrets`` silently skips a
1291 # ``None`` secret. Drop the traceback chain.
1292 safe_msg = redact_secrets(str(e), password)
1293 logger.warning(
1294 f"[DOC_SCHEDULER] Error processing documents for user {username}: {safe_msg}"
1295 )
1297 @thread_cleanup
1298 def _reconcile_unindexed_documents(self, username: str) -> None:
1299 """Unified background reconciler that indexes ANY unindexed document.
1301 Self-healing follow-up to the immediate auto-index queue (PR #3939),
1302 which caps the queue and DROPS documents on saturation, AND replacement
1303 for the retired research-scoped ``generate_rag`` indexing block that
1304 used to live inline in ``_process_user_documents``. A single scheduled
1305 job now covers every unindexed document, so library uploads and research
1306 downloads can no longer be permanently missed.
1308 Two cases are handled per tick, each with its OWN independent
1309 ``_LIBRARY_SWEEP_BATCH`` budget so the work done in a single thread tick
1310 stays bounded (total <= 2 x ``_LIBRARY_SWEEP_BATCH``). The budgets are
1311 decoupled on purpose: case (a) only marks a row indexed on SUCCESS, so a
1312 block of permanently-failing case-(a) rows must not be able to consume
1313 case (b)'s budget and starve the research-orphan path:
1315 (a) In-collection unindexed: documents that already have a
1316 ``DocumentCollection`` link (e.g. manual uploads — ``upload_to_
1317 collection`` always creates the row) with ``indexed`` False and
1318 text content. Indexed via the per-collection RAG factory so each
1319 collection's own embedding config is honored.
1320 (b) Research orphans: ``Document`` rows with ``research_id`` set and
1321 text content that have NO ``DocumentCollection`` link in the default
1322 library collection yet (research downloads that were never
1323 ingested). ``index_document(doc_id, default_library_id, ...)`` calls
1324 ``ensure_in_collection`` internally, so it ingests + indexes in one
1325 call.
1327 Behaviour:
1328 - Gated by EITHER ``sweep_library_collections`` OR ``generate_rag`` —
1329 the ``generate_rag`` arm preserves the legacy "index research
1330 downloads" behaviour. Off by default; early-returns when neither set.
1331 - Idempotent: uses ``index_document(..., force_reindex=False)`` and only
1332 selects rows that are not yet indexed, so already-indexed documents
1333 are never touched.
1334 - Self-rate-limited: each case is capped at ``_LIBRARY_SWEEP_BATCH``
1335 documents per tick (so total <= 2 x ``_LIBRARY_SWEEP_BATCH``); the job
1336 is scheduled with ``max_instances=1``.
1337 """
1338 logger.info(
1339 f"[RECONCILER] Starting unindexed-document reconcile for user {username}"
1340 )
1342 # Pre-declared so the except handlers can pass it to redact_secrets
1343 # even if the retrieve() call below itself raises.
1344 password = None
1345 try:
1346 session_info = self.user_sessions.get(username)
1347 if not session_info:
1348 logger.warning(
1349 f"[RECONCILER] No session info found for user {username}"
1350 )
1351 return
1353 password = self._credential_store.retrieve(username)
1354 if not password:
1355 logger.warning(
1356 f"[RECONCILER] Credentials expired for user {username}"
1357 )
1358 return
1360 # Get user's document scheduler settings (cached).
1361 settings = self._get_document_scheduler_settings(username)
1363 # Gate at runtime too, not just at scheduling: the already-live
1364 # APScheduler job keeps firing after the document scheduler is
1365 # disabled until the next reschedule, and the setting description
1366 # promises the sweep only runs while the scheduler is enabled.
1367 # OFF by default: runs only when the scheduler is enabled AND
1368 # EITHER opt-in is set. The generate_rag arm preserves the legacy
1369 # research-download indexing behaviour from _process_user_documents.
1370 if not settings.enabled or not (
1371 settings.sweep_library_collections or settings.generate_rag
1372 ):
1373 logger.debug(
1374 f"[RECONCILER] Indexing disabled for user {username}"
1375 )
1376 return
1378 # Lazy import of the RAG factory. Imported here (not at module
1379 # top) to keep the import surface of this scheduler module small
1380 # and consistent with the other lazy imports in this file; the
1381 # factory itself has no import dependency on the scheduler so a
1382 # top-level import would also be safe.
1383 from ..research_library.services.rag_service_factory import (
1384 get_rag_service,
1385 )
1387 from ..database.session_context import get_user_db_session
1388 from ..settings.manager import SettingsManager
1390 with get_user_db_session(username, password) as db:
1391 settings_manager = SettingsManager(db)
1393 # Arm the PEP-578 audit-hook backstop for this scheduled run,
1394 # mirroring _process_user_documents. Indexing itself doesn't
1395 # download, but embedding providers may make network calls;
1396 # this keeps defense-in-depth parity with an interactive run.
1397 # Cleared by @thread_cleanup on exit.
1398 if not self._arm_egress_backstop(settings_manager, username):
1399 return
1401 # Resolve the parallel-worker count from the user's saved
1402 # settings so a configured ``rag.indexing_max_parallel_docs``
1403 # of 1 truly restores the legacy strictly-sequential sweep
1404 # (the SSE routes also resolve this from settings, but the
1405 # reconciler runs as a scheduled background thread without a
1406 # Flask app context, so it uses the already-bound db session
1407 # via ``SettingsManager(db)`` rather than the global
1408 # ``get_settings_manager()`` helper — which the pre-commit
1409 # ``check-settings-manager-thread-safety`` hook forbids inside
1410 # ``@thread_cleanup``-decorated functions without an explicit
1411 # ``db_session=`` argument).
1412 try:
1413 _max_parallel = int(
1414 settings_manager.get_setting(
1415 "rag.indexing_max_parallel_docs", 4
1416 )
1417 )
1418 except Exception:
1419 _max_parallel = 4
1420 _max_parallel = max(1, min(_max_parallel, 16))
1422 total_indexed = 0
1424 # ---- Case (a): in-collection unindexed documents ----------
1425 # Bounded by _LIBRARY_SWEEP_BATCH so a large backlog self-heals
1426 # over successive ticks. Only rows with actual text content can
1427 # be indexed. RANDOMIZED selection (not a stable id order): a
1428 # row leaves this candidate set only on SUCCESS and we track no
1429 # per-row failure state, so a deterministic order would let a
1430 # block of permanently-failing low-id rows (e.g. empty-text /
1431 # scanned PDFs that always return an indexing error yet pass the
1432 # text_content IS NOT NULL filter) win the LIMIT slots every
1433 # tick and starve indexable higher-id rows forever. Random
1434 # sampling gives every indexable row a chance each tick, so
1435 # progress is eventually made despite a permanent-failure set.
1436 unindexed = (
1437 db.query(
1438 DocumentCollection.document_id,
1439 DocumentCollection.collection_id,
1440 )
1441 .join(
1442 Document,
1443 Document.id == DocumentCollection.document_id,
1444 )
1445 .filter(
1446 DocumentCollection.indexed.is_(False),
1447 Document.text_content.isnot(None),
1448 )
1449 .order_by(func.random())
1450 .limit(_LIBRARY_SWEEP_BATCH)
1451 .all()
1452 )
1454 # Group document ids by collection so we build exactly one RAG
1455 # service per collection (each collection can have its own
1456 # embedding config).
1457 docs_by_collection: Dict[str, List[str]] = {}
1458 for doc_id, coll_id in unindexed:
1459 docs_by_collection.setdefault(coll_id, []).append(doc_id)
1461 if unindexed:
1462 logger.info(
1463 f"[RECONCILER] Found {len(unindexed)} in-collection "
1464 f"unindexed document(s) across "
1465 f"{len(docs_by_collection)} collection(s) for {username}"
1466 )
1468 for coll_id, doc_ids in docs_by_collection.items():
1469 try:
1470 # USE THE FACTORY so per-collection embedding settings
1471 # (model/provider/chunking/etc.) stored on the
1472 # collection are honored — get_rag_service loads them
1473 # from the collection row when collection_id is given.
1474 with get_rag_service(
1475 username,
1476 collection_id=coll_id,
1477 db_password=password,
1478 ) as rag_service:
1479 # Fan out across the per-collection batch using
1480 # the user-configured parallel-worker count
1481 # resolved above. Clamped to ``[1, 16]`` so a
1482 # misconfigured value can't blow the pool size
1483 # past the documented range.
1484 doc_info = [(doc_id, "") for doc_id in doc_ids]
1485 aggregate = rag_service.index_documents_parallel(
1486 doc_info,
1487 coll_id,
1488 force_reindex=False,
1489 max_workers=_max_parallel,
1490 )
1491 for doc_id in doc_ids:
1492 result = aggregate["results"].get(doc_id, {})
1493 if result.get("status") == "success":
1494 total_indexed += 1
1495 logger.debug(
1496 f"[RECONCILER] Indexed document {doc_id} "
1497 f"into collection {coll_id} with "
1498 f"{result.get('chunk_count', 0)} chunks"
1499 )
1500 elif result.get("status") == "error": 1500 ↛ 1491line 1500 didn't jump to line 1491 because the condition on line 1500 was always true
1501 # ``password`` is in scope and was
1502 # passed into ``get_rag_service``.
1503 # Drop the traceback chain + redact
1504 # str(e) to avoid leaking the
1505 # SQLCipher master password.
1506 safe_msg = redact_secrets(
1507 str(result.get("error", "")),
1508 password,
1509 )
1510 logger.warning(
1511 f"[RECONCILER] Failed to index document "
1512 f"{doc_id} into collection {coll_id}: {safe_msg}"
1513 )
1514 except Exception as coll_error:
1515 # Recover the shared thread-local session before moving
1516 # on to the next collection so its queries don't run on
1517 # a poisoned session.
1518 safe_rollback(db, "RECONCILER collection")
1519 safe_msg = redact_secrets(str(coll_error), password)
1520 logger.warning(
1521 f"[RECONCILER] Failed to index collection {coll_id}: {safe_msg}"
1522 )
1524 # ---- Case (b): research orphans -> default library ---------
1525 # Research downloads land as Document rows (research_id set)
1526 # with NO DocumentCollection link yet. index_document()
1527 # ensure_in_collection's the default-library link, so this
1528 # ingests + indexes in one call.
1529 #
1530 # Case (b) gets its OWN independent _LIBRARY_SWEEP_BATCH budget
1531 # rather than the leftover of case (a). Case (a) only flips a
1532 # row to indexed=True on SUCCESS, so a block of permanently
1533 # failing case-(a) rows (empty text, embedding/FAISS errors,
1534 # PolicyDeniedError under egress denial) would otherwise fill
1535 # the LIMIT every tick, leave the leftover at 0, and starve the
1536 # research-orphan path forever — regressing the no-regression
1537 # promise for generate_rag-only users. Decoupling the budgets
1538 # caps total work at 2 x _LIBRARY_SWEEP_BATCH per tick, which is
1539 # acceptable. RANDOMIZED selection (see case (a)) so a block of
1540 # permanently-failing low-id orphans can't pin the LIMIT slots
1541 # every tick and starve the rest.
1542 #
1543 # Resolve the default library collection once.
1544 default_library_id = get_default_library_id(username, password)
1546 orphans = (
1547 db.query(Document.id)
1548 .outerjoin(
1549 DocumentCollection,
1550 (DocumentCollection.document_id == Document.id)
1551 & (
1552 DocumentCollection.collection_id
1553 == default_library_id
1554 ),
1555 )
1556 .filter(
1557 Document.research_id.isnot(None),
1558 Document.text_content.isnot(None),
1559 DocumentCollection.id.is_(None),
1560 )
1561 .order_by(func.random())
1562 .limit(_LIBRARY_SWEEP_BATCH)
1563 .all()
1564 )
1566 if orphans:
1567 logger.info(
1568 f"[RECONCILER] Found {len(orphans)} research "
1569 f"orphan document(s) to ingest into the default "
1570 f"library for {username}"
1571 )
1572 try:
1573 with get_rag_service(
1574 username,
1575 collection_id=default_library_id,
1576 db_password=password,
1577 ) as rag_service:
1578 # Same parallel fan-out as the in-collection
1579 # branch above; ``_max_parallel`` was resolved
1580 # from the user's saved
1581 # ``rag.indexing_max_parallel_docs`` setting at
1582 # the top of this tick.
1583 orphan_ids = [doc_id for (doc_id,) in orphans]
1584 doc_info = [(doc_id, "") for doc_id in orphan_ids]
1585 aggregate = rag_service.index_documents_parallel(
1586 doc_info,
1587 default_library_id,
1588 force_reindex=False,
1589 max_workers=_max_parallel,
1590 )
1591 for doc_id in orphan_ids:
1592 result = aggregate["results"].get(doc_id, {})
1593 if result.get("status") == "success": 1593 ↛ 1602line 1593 didn't jump to line 1602 because the condition on line 1593 was always true
1594 total_indexed += 1
1595 logger.debug(
1596 f"[RECONCILER] Ingested + "
1597 f"indexed research orphan "
1598 f"{doc_id} into the default "
1599 f"library with "
1600 f"{result.get('chunk_count', 0)} chunks"
1601 )
1602 elif result.get("status") == "error":
1603 safe_msg = redact_secrets(
1604 str(result.get("error", "")),
1605 password,
1606 )
1607 logger.warning(
1608 f"[RECONCILER] Failed to index "
1609 f"research orphan {doc_id}: {safe_msg}"
1610 )
1611 except Exception as orphan_error:
1612 safe_rollback(db, "RECONCILER orphans")
1613 safe_msg = redact_secrets(str(orphan_error), password)
1614 logger.warning(
1615 f"[RECONCILER] Failed to index research "
1616 f"orphans into the default library: {safe_msg}"
1617 )
1619 logger.info(
1620 f"[RECONCILER] Completed reconcile for user {username}: "
1621 f"{total_indexed} document(s) indexed "
1622 f"(per-case batch cap {_LIBRARY_SWEEP_BATCH})"
1623 )
1625 except Exception as e:
1626 # ``password`` is pre-declared as ``None`` at the top of the
1627 # function, so it is always bound here even if the retrieve()
1628 # call itself raised. ``redact_secrets`` silently skips a
1629 # ``None`` secret. Drop the traceback chain.
1630 safe_msg = redact_secrets(str(e), password)
1631 logger.warning(
1632 f"[RECONCILER] Error reconciling unindexed documents for user {username}: {safe_msg}"
1633 )
1635 def get_document_scheduler_status(self, username: str) -> Dict[str, Any]:
1636 """Get document scheduler status for a specific user."""
1637 try:
1638 session_info = self.user_sessions.get(username)
1639 if not session_info:
1640 return {
1641 "enabled": False,
1642 "message": "User not found in scheduler",
1643 }
1645 # Get user's document scheduler settings (cached)
1646 settings = self._get_document_scheduler_settings(username)
1648 # Check if user has document processing job
1649 job_id = f"{username}_document_processing"
1650 has_job = job_id in session_info.get("scheduled_jobs", set())
1652 return {
1653 "enabled": settings.enabled,
1654 "interval_seconds": settings.interval_seconds,
1655 "processing_options": {
1656 "download_pdfs": settings.download_pdfs,
1657 "extract_text": settings.extract_text,
1658 # generate_rag and sweep_library_collections both gate the
1659 # unified reconciler (_reconcile_unindexed_documents).
1660 "generate_rag": settings.generate_rag,
1661 "sweep_library_collections": settings.sweep_library_collections,
1662 },
1663 "last_run": settings.last_run,
1664 "has_scheduled_job": has_job,
1665 "user_active": username in self.user_sessions,
1666 }
1668 except Exception as e:
1669 # No ``password`` local in this method, but caller frames
1670 # (e.g. a route handler that already retrieved the user's
1671 # password) could be rendered under loguru ``diagnose=True``.
1672 # Drop the traceback by using ``logger.warning`` without
1673 # ``exc_info``.
1674 safe_msg = redact_secrets(str(e), None)
1675 logger.warning(
1676 f"Error getting document scheduler status for user {username}: {safe_msg}"
1677 )
1678 return {
1679 "enabled": False,
1680 "message": f"Failed to retrieve scheduler status: {type(e).__name__}",
1681 }
1683 def trigger_document_processing(self, username: str) -> bool:
1684 """Trigger immediate document processing for a user."""
1685 logger.info(
1686 f"[DOC_SCHEDULER] Manual trigger requested for user {username}"
1687 )
1688 try:
1689 session_info = self.user_sessions.get(username)
1690 if not session_info:
1691 logger.warning(
1692 f"[DOC_SCHEDULER] User {username} not found in scheduler"
1693 )
1694 logger.debug(
1695 f"[DOC_SCHEDULER] Available users: {list(self.user_sessions.keys())}"
1696 )
1697 return False
1699 if not self.is_running:
1700 logger.warning(
1701 f"[DOC_SCHEDULER] Scheduler not running, cannot trigger document processing for {username}"
1702 )
1703 return False
1705 # Trigger immediate processing
1706 job_id = f"{username}_document_processing_manual"
1707 logger.debug(f"[DOC_SCHEDULER] Scheduling manual job {job_id}")
1709 self.scheduler.add_job(
1710 func=self._wrap_job(
1711 self._process_user_documents, username=username
1712 ),
1713 args=[username],
1714 trigger="date",
1715 run_date=datetime.now(UTC) + timedelta(seconds=1),
1716 id=job_id,
1717 name=f"Manual Document Processing for {username}",
1718 replace_existing=True,
1719 )
1721 # Verify job was added
1722 job = self.scheduler.get_job(job_id)
1723 if job:
1724 logger.info(
1725 f"[DOC_SCHEDULER] Successfully triggered manual document processing for user {username}, job {job_id}, next run: {job.next_run_time}"
1726 )
1727 else:
1728 logger.error(
1729 f"[DOC_SCHEDULER] Failed to verify manual job {job_id} was added!"
1730 )
1731 return False
1733 return True
1735 except Exception as e:
1736 # No ``password`` local in this method, but caller frames
1737 # could hold one — drop the traceback to avoid frame-local
1738 # rendering under ``diagnose=True``.
1739 safe_msg = redact_secrets(str(e), None)
1740 logger.warning(
1741 f"[DOC_SCHEDULER] Error triggering document processing for user {username}: {safe_msg}"
1742 )
1743 return False
1745 # -- Zotero auto-sync -------------------------------------------------
1747 def _schedule_zotero_sync(self, username: str):
1748 """Schedule (or refresh) the Zotero auto-sync job for a user.
1750 No-op unless the user has enabled the Zotero integration *and*
1751 background auto-sync. Any existing job is removed first so toggling
1752 the setting off actually unschedules it.
1753 """
1754 # Pre-declared so the except handler can pass it to redact_secrets
1755 # even if retrieve()/get_config() raises — get_config() opens the
1756 # encrypted DB, so an error message could embed the SQLCipher key.
1757 password = None
1758 try:
1759 session_info = self.user_sessions.get(username)
1760 if not session_info: 1760 ↛ 1761line 1760 didn't jump to line 1761 because the condition on line 1760 was never true
1761 return
1763 job_id = f"{username}_zotero_sync"
1765 def _drop_existing_job():
1766 try:
1767 self.scheduler.remove_job(job_id)
1768 session_info["scheduled_jobs"].discard(job_id)
1769 except JobLookupError:
1770 pass
1772 password = self._credential_store.retrieve(username)
1773 if not password:
1774 # Expired credentials: the job could never run anyway, and
1775 # disabling auto-sync in this state must still unschedule it —
1776 # otherwise it keeps ticking (and logging a warning) until
1777 # logout.
1778 _drop_existing_job()
1779 return
1781 from ..research_library.zotero import ZoteroSyncService
1783 cfg = ZoteroSyncService(username, password).get_config()
1785 # Remove only AFTER the config read succeeded: a transient error
1786 # from the encrypted-DB read must leave a healthy existing job in
1787 # place (it would otherwise stay gone until the next login).
1788 # Accepted trade-off: if that error races a just-disabled toggle,
1789 # the stale job stays scheduled — but it is inert, because
1790 # _sync_user_zotero re-checks the config at every tick and no-ops
1791 # when auto-sync is disabled.
1792 _drop_existing_job()
1794 if not (cfg.is_configured and cfg.auto_sync_enabled):
1795 logger.debug(
1796 f"[ZOTERO_SCHEDULER] Auto-sync not enabled for {username}"
1797 )
1798 return
1800 interval_minutes = max(15, int(cfg.sync_interval_minutes or 360))
1801 self.scheduler.add_job(
1802 func=self._wrap_job(self._sync_user_zotero, username=username),
1803 args=[username],
1804 trigger="interval",
1805 minutes=interval_minutes,
1806 id=job_id,
1807 name=f"Zotero Sync for {username}",
1808 jitter=60,
1809 max_instances=1,
1810 replace_existing=True,
1811 )
1812 session_info["scheduled_jobs"].add(job_id)
1813 logger.info(
1814 f"[ZOTERO_SCHEDULER] Scheduled Zotero sync for {username} "
1815 f"every {interval_minutes} minutes"
1816 )
1817 except Exception as e:
1818 safe_msg = redact_secrets(str(e), password)
1819 logger.warning(
1820 f"[ZOTERO_SCHEDULER] Error scheduling Zotero sync for "
1821 f"{username}: {safe_msg}"
1822 )
1824 @thread_cleanup
1825 def _sync_user_zotero(self, username: str):
1826 """Background job: run a Zotero sync for a user."""
1827 password = None
1828 try:
1829 session_info = self.user_sessions.get(username)
1830 if not session_info:
1831 return
1833 password = self._credential_store.retrieve(username)
1834 if not password:
1835 logger.warning(
1836 f"[ZOTERO_SCHEDULER] Credentials expired for {username}"
1837 )
1838 return
1840 from ..research_library.zotero import ZoteroSyncService
1842 service = ZoteroSyncService(username, password)
1843 cfg = service.get_config()
1844 if not (cfg.is_configured and cfg.auto_sync_enabled):
1845 logger.debug(
1846 f"[ZOTERO_SCHEDULER] Auto-sync not enabled for {username}"
1847 )
1848 return
1850 logger.info(
1851 f"[ZOTERO_SCHEDULER] Running Zotero sync for {username}"
1852 )
1853 result = service.sync_all()
1854 logger.info(
1855 f"[ZOTERO_SCHEDULER] Zotero sync for {username}: "
1856 f"imported={result.get('imported')}, "
1857 f"updated={result.get('updated')}, "
1858 f"removed={result.get('removed')}, "
1859 f"skipped={result.get('skipped')}, "
1860 f"errors={result.get('errors')}"
1861 )
1862 except Exception as e:
1863 # ``password`` was retrieved above and passed into the sync
1864 # service (which opens the encrypted DB). Drop the traceback
1865 # chain and redact str(e) so the SQLCipher master password
1866 # cannot leak under loguru ``diagnose=True``.
1867 safe_msg = redact_secrets(str(e), password)
1868 logger.warning(
1869 f"[ZOTERO_SCHEDULER] Error syncing Zotero for {username}: "
1870 f"{safe_msg}"
1871 )
1873 @thread_cleanup
1874 def _check_user_overdue_subscriptions(self, username: str):
1875 """Check and immediately run any overdue subscriptions for a user."""
1876 # Pre-declared so the except handler can pass it to redact_secrets
1877 # even if the retrieve() call below itself raises.
1878 password = None
1879 try:
1880 session_info = self.user_sessions.get(username)
1881 if not session_info:
1882 return
1884 password = self._credential_store.retrieve(username)
1885 if not password:
1886 return
1888 # Get user's overdue subscriptions
1889 from ..database.session_context import get_user_db_session
1890 from ..database.models.news import NewsSubscription
1891 from datetime import timezone
1893 with get_user_db_session(username, password) as db:
1894 now = datetime.now(timezone.utc)
1895 overdue_subs = (
1896 db.query(NewsSubscription)
1897 .filter(NewsSubscription.due_filter(now))
1898 .all()
1899 )
1901 if overdue_subs:
1902 logger.info(
1903 f"Found {len(overdue_subs)} overdue subscriptions for {username}"
1904 )
1906 for sub in overdue_subs:
1907 # Run immediately with small random delay
1908 # Security: random delay to stagger overdue jobs, not security-sensitive
1909 delay_seconds = random.randint(1, 30)
1910 job_id = (
1911 f"overdue_{username}_{sub.id}_{int(now.timestamp())}"
1912 )
1914 self.scheduler.add_job(
1915 func=self._wrap_job(
1916 self._check_subscription, username=username
1917 ),
1918 args=[username, sub.id],
1919 trigger="date",
1920 run_date=now + timedelta(seconds=delay_seconds),
1921 id=job_id,
1922 name=f"Overdue: {sub.name or sub.query_or_topic[:30]}",
1923 replace_existing=True,
1924 )
1926 logger.info(
1927 f"Scheduled overdue subscription {sub.id} to run in {delay_seconds} seconds"
1928 )
1930 except Exception as e:
1931 # ``password`` was retrieved above and passed into
1932 # ``get_user_db_session``. Drop traceback + redact str(e)
1933 # to avoid leaking the SQLCipher master password.
1934 safe_msg = redact_secrets(str(e), password)
1935 logger.warning(
1936 f"Error checking overdue subscriptions for {username}: {safe_msg}"
1937 )
1939 @thread_cleanup
1940 def _check_subscription(self, username: str, subscription_id: int):
1941 """Check and refresh a single subscription."""
1942 logger.info(
1943 f"_check_subscription called for user {username}, subscription {subscription_id}"
1944 )
1945 # Pre-declared so the except handler can pass it to redact_secrets
1946 # even if the retrieve() call below itself raises.
1947 password = None
1948 try:
1949 session_info = self.user_sessions.get(username)
1950 if not session_info:
1951 # User no longer active, cancel job
1952 job_id = f"{username}_{subscription_id}"
1953 try:
1954 self.scheduler.remove_job(job_id)
1955 except JobLookupError:
1956 pass
1957 return
1959 password = self._credential_store.retrieve(username)
1960 if not password: 1960 ↛ 1961line 1960 didn't jump to line 1961 because the condition on line 1960 was never true
1961 logger.warning(
1962 f"Credentials expired for {username}, skipping subscription check"
1963 )
1964 return
1966 # Get subscription details
1967 from ..database.session_context import get_user_db_session
1968 from ..database.models.news import (
1969 NewsSubscription,
1970 SubscriptionStatus,
1971 )
1972 from ..news.subscription_runner import advance_refresh_schedule
1974 with get_user_db_session(username, password) as db:
1975 sub = db.get(NewsSubscription, subscription_id)
1976 if not sub or sub.status != SubscriptionStatus.ACTIVE.value:
1977 logger.info(
1978 f"Subscription {subscription_id} not active, skipping"
1979 )
1980 return
1982 # Prepare query with date replacement using user's timezone
1983 query = sub.query_or_topic
1984 if "YYYY-MM-DD" in query:
1985 from local_deep_research.news.core.utils import (
1986 get_local_date_string,
1987 )
1988 from ..settings.manager import SettingsManager
1990 settings_manager = SettingsManager(db)
1991 local_date = get_local_date_string(settings_manager)
1992 query = query.replace("YYYY-MM-DD", local_date)
1994 # Update last/next refresh times
1995 advance_refresh_schedule(sub, datetime.now(UTC))
1996 db.commit()
1998 subscription_data = {
1999 "id": sub.id,
2000 "name": sub.name,
2001 "query": query,
2002 "original_query": sub.query_or_topic,
2003 "model_provider": sub.model_provider,
2004 "model": sub.model,
2005 "search_strategy": sub.search_strategy,
2006 "search_engine": sub.search_engine,
2007 }
2009 logger.info(
2010 f"Refreshing subscription {subscription_id}: {subscription_data['name']}"
2011 )
2013 # Trigger research synchronously using requests with proper auth
2014 self._trigger_subscription_research_sync(
2015 username, subscription_data
2016 )
2018 # Reschedule for next interval if using interval trigger
2019 job_id = f"{username}_{subscription_id}"
2020 job = self.scheduler.get_job(job_id)
2021 if job and job.trigger.__class__.__name__ == "DateTrigger":
2022 # For date triggers, reschedule
2023 # Security: random jitter to distribute subscription timing, not security-sensitive
2024 next_run = datetime.now(UTC) + timedelta(
2025 minutes=sub.refresh_interval_minutes,
2026 seconds=random.randint(
2027 0, int(self.config.get("max_jitter_seconds", 300))
2028 ),
2029 )
2030 self.scheduler.add_job(
2031 func=self._wrap_job(
2032 self._check_subscription, username=username
2033 ),
2034 args=[username, subscription_id],
2035 trigger="date",
2036 run_date=next_run,
2037 id=job_id,
2038 replace_existing=True,
2039 )
2041 except Exception as e:
2042 # ``password`` was retrieved above and passed into
2043 # ``get_user_db_session``. Drop traceback + redact str(e)
2044 # to avoid leaking the SQLCipher master password.
2045 safe_msg = redact_secrets(str(e), password)
2046 logger.warning(
2047 f"Error checking subscription {subscription_id}: {safe_msg}"
2048 )
2050 @thread_cleanup
2051 def _trigger_subscription_research_sync(
2052 self, username: str, subscription: Dict[str, Any]
2053 ):
2054 """Trigger research for a subscription using programmatic API."""
2055 from ..config.thread_settings import set_settings_context
2057 # Pre-declared so the except handler can pass it to redact_secrets
2058 # even if the retrieve() call below itself raises.
2059 password = None
2060 try:
2061 # Get user's password from session info
2062 session_info = self.user_sessions.get(username)
2063 if not session_info:
2064 logger.error(f"No session info for user {username}")
2065 return
2067 password = self._credential_store.retrieve(username)
2068 if not password: 2068 ↛ 2069line 2068 didn't jump to line 2069 because the condition on line 2068 was never true
2069 logger.error(f"Credentials expired for user {username}")
2070 return
2072 # Generate research ID
2073 import uuid
2075 research_id = str(uuid.uuid4())
2077 logger.info(
2078 f"Starting research {research_id} for subscription {subscription['id']}"
2079 )
2081 # Get user settings for research
2082 from ..database.session_context import get_user_db_session
2083 from ..settings.manager import SettingsManager
2085 with get_user_db_session(username, password) as db:
2086 settings_manager = SettingsManager(db)
2087 settings_snapshot = settings_manager.get_settings_snapshot()
2089 # Use the search engine from the subscription if specified
2090 search_engine = subscription.get("search_engine")
2092 if search_engine:
2093 settings_snapshot["search.tool"] = {
2094 "value": search_engine,
2095 "ui_element": "select",
2096 }
2097 logger.info(
2098 f"Using subscription's search engine: '{search_engine}' for {subscription['id']}"
2099 )
2100 else:
2101 # Use the user's default search tool from their settings
2102 default_search_tool = settings_snapshot.get(
2103 "search.tool", DEFAULT_SEARCH_TOOL
2104 )
2105 logger.info(
2106 f"Using user's default search tool: '{default_search_tool}' for {subscription['id']}"
2107 )
2109 logger.debug(
2110 f"Settings snapshot has {len(settings_snapshot)} settings"
2111 )
2112 # Log a few key settings to verify they're present
2113 logger.debug(
2114 f"Key settings: llm.model={settings_snapshot.get('llm.model')}, llm.provider={settings_snapshot.get('llm.provider')}, search.tool={settings_snapshot.get('search.tool')}"
2115 )
2117 # Set up research parameters
2118 query = subscription["query"]
2120 # Build metadata for news search
2121 metadata = {
2122 "is_news_search": True,
2123 "search_type": "news_analysis",
2124 "display_in": "news_feed",
2125 "subscription_id": subscription["id"],
2126 "triggered_by": "scheduler",
2127 "subscription_name": subscription["name"],
2128 "title": subscription["name"] if subscription["name"] else None,
2129 "scheduled_at": datetime.now(UTC).isoformat(),
2130 "original_query": subscription["original_query"],
2131 "user_id": username,
2132 }
2134 # Use programmatic API with settings context
2135 from ..api.research_functions import quick_summary
2137 # Create and set settings context for this thread
2138 settings_context = SnapshotSettingsContext(settings_snapshot)
2139 set_settings_context(settings_context)
2141 # Get search strategy from subscription data
2142 search_strategy = subscription.get("search_strategy")
2144 # Build kwargs for quick_summary, only including
2145 # search_strategy if the subscription specifies one.
2146 quick_summary_kwargs = {
2147 "query": query,
2148 "research_id": research_id,
2149 "username": username,
2150 "user_password": password,
2151 "settings_snapshot": settings_snapshot,
2152 "model_name": subscription.get("model"),
2153 "provider": subscription.get("model_provider"),
2154 "metadata": metadata,
2155 "search_original_query": False, # Don't send long subscription prompts to search engines
2156 }
2157 if search_strategy: 2157 ↛ 2160line 2157 didn't jump to line 2160 because the condition on line 2157 was always true
2158 quick_summary_kwargs["search_strategy"] = search_strategy
2160 result = quick_summary(**quick_summary_kwargs)
2162 logger.info(
2163 f"Completed research {research_id} for subscription {subscription['id']}"
2164 )
2166 # Store the research result in the database
2167 self._store_research_result(
2168 username,
2169 password,
2170 research_id,
2171 subscription["id"],
2172 result,
2173 subscription,
2174 )
2176 except Exception as e:
2177 # ``password`` was retrieved from the credential store at
2178 # the top of this function and passed through to
2179 # ``get_user_db_session``, ``quick_summary``
2180 # (``user_password``), and ``_store_research_result``. A
2181 # SQLAlchemy / requests exception from any of those paths
2182 # could carry frame locals containing the SQLCipher master
2183 # password — drop the traceback chain and redact str(e).
2184 safe_msg = redact_secrets(str(e), password)
2185 logger.warning(
2186 f"Error triggering research for subscription {subscription['id']}: {safe_msg}"
2187 )
2189 def _store_research_result(
2190 self,
2191 username: str,
2192 password: str,
2193 research_id: str,
2194 subscription_id: int,
2195 result: Dict[str, Any],
2196 subscription: Dict[str, Any],
2197 ):
2198 """Store research result in database for news display."""
2199 try:
2200 from ..database.session_context import get_user_db_session
2201 from ..database.models import ResearchHistory
2202 from ..settings.manager import SettingsManager
2203 import json
2205 # Convert result to JSON-serializable format
2206 def make_serializable(obj):
2207 """Convert non-serializable objects to dictionaries."""
2208 if hasattr(obj, "dict"):
2209 return obj.dict()
2210 if hasattr(obj, "__dict__"): 2210 ↛ 2211line 2210 didn't jump to line 2211 because the condition on line 2210 was never true
2211 return {
2212 k: make_serializable(v)
2213 for k, v in obj.__dict__.items()
2214 if not k.startswith("_")
2215 }
2216 if isinstance(obj, (list, tuple)):
2217 return [make_serializable(item) for item in obj]
2218 if isinstance(obj, dict):
2219 return {k: make_serializable(v) for k, v in obj.items()}
2220 return obj
2222 serializable_result = make_serializable(result)
2224 with get_user_db_session(username, password) as db:
2225 # Get user settings to store in metadata
2226 settings_manager = SettingsManager(db)
2227 settings_snapshot = settings_manager.get_settings_snapshot()
2228 # Scope the snapshot to this subscription's owner so the
2229 # post-hoc headline/topic LLM calls resolve a per-user
2230 # provider and honor local-only policy. get_settings_snapshot
2231 # omits `_username`; without it the news findings (which can
2232 # come from a private-retriever run) could fail open to a
2233 # cloud LLM.
2234 from ..search_system import ensure_snapshot_username
2236 settings_snapshot = ensure_snapshot_username(
2237 settings_snapshot, username
2238 )
2240 # Get the report content - check both 'report' and 'summary' fields
2241 report_content = serializable_result.get(
2242 "report"
2243 ) or serializable_result.get("summary")
2244 logger.debug(
2245 f"Report content length: {len(report_content) if report_content else 0} chars"
2246 )
2248 # Extract sources/links from the result. They get
2249 # persisted to research_resources AFTER history_entry
2250 # commits below (FK requires research_id to exist).
2251 sources = serializable_result.get("sources", [])
2253 # Then format citations in the report content
2254 if report_content:
2255 # Import citation formatter
2256 from ..text_optimization.citation_formatter import (
2257 CitationFormatter,
2258 CitationMode,
2259 )
2260 from ..config.search_config import (
2261 get_setting_from_snapshot,
2262 )
2264 # Get citation format from settings
2265 citation_format = get_setting_from_snapshot(
2266 "report.citation_format", "domain_id_hyperlinks"
2267 )
2268 mode_map = {
2269 "number_hyperlinks": CitationMode.NUMBER_HYPERLINKS,
2270 "domain_hyperlinks": CitationMode.DOMAIN_HYPERLINKS,
2271 "domain_id_hyperlinks": CitationMode.DOMAIN_ID_HYPERLINKS,
2272 "domain_id_always_hyperlinks": CitationMode.DOMAIN_ID_ALWAYS_HYPERLINKS,
2273 "source_tagged_hyperlinks": CitationMode.SOURCE_TAGGED_HYPERLINKS,
2274 "no_hyperlinks": CitationMode.NO_HYPERLINKS,
2275 }
2276 mode = mode_map.get(
2277 citation_format, CitationMode.DOMAIN_ID_HYPERLINKS
2278 )
2279 formatter = CitationFormatter(mode=mode)
2281 # Format citations within the content
2282 report_content = formatter.format_document(report_content)
2284 if not report_content:
2285 # If neither field exists, use the full result as JSON
2286 report_content = json.dumps(serializable_result)
2288 # Generate headline and topics for news searches
2289 from ..news.utils.headline_generator import generate_headline
2290 from ..news.utils.topic_generator import generate_topics
2292 query_text = result.get(
2293 "query", subscription.get("query", "News Update")
2294 )
2296 # Generate headline from the actual research findings
2297 logger.info(
2298 f"Generating headline for subscription {subscription_id}"
2299 )
2300 generated_headline = generate_headline(
2301 query=query_text,
2302 findings=report_content,
2303 max_length=200, # Allow longer headlines for news
2304 settings_snapshot=settings_snapshot,
2305 )
2307 # Generate topics from the findings
2308 logger.info(
2309 f"Generating topics for subscription {subscription_id}"
2310 )
2311 generated_topics = generate_topics(
2312 query=query_text,
2313 findings=report_content,
2314 category=subscription.get("name", "News"),
2315 max_topics=6,
2316 settings_snapshot=settings_snapshot,
2317 )
2319 logger.info(
2320 f"Generated headline: {generated_headline}, topics: {generated_topics}"
2321 )
2323 # Get subscription name for metadata
2324 subscription_name = subscription.get("name", "")
2326 # Use generated headline as title, or fallback
2327 if generated_headline:
2328 title = generated_headline
2329 else:
2330 if subscription_name:
2331 title = f"{subscription_name} - {datetime.now(UTC).isoformat(timespec='minutes')}"
2332 else:
2333 title = f"{query_text[:60]}... - {datetime.now(UTC).isoformat(timespec='minutes')}"
2335 # Create research history entry
2336 history_entry = ResearchHistory(
2337 id=research_id,
2338 query=result.get("query", ""),
2339 mode="news_subscription",
2340 status="completed",
2341 created_at=datetime.now(UTC).isoformat(),
2342 completed_at=datetime.now(UTC).isoformat(),
2343 title=title,
2344 research_meta={
2345 "subscription_id": subscription_id,
2346 "triggered_by": "scheduler",
2347 "is_news_search": True,
2348 "username": username,
2349 "subscription_name": subscription_name, # Store subscription name for display
2350 "settings_snapshot": settings_snapshot, # Store settings snapshot for later retrieval
2351 "generated_headline": generated_headline, # Store generated headline for news display
2352 "generated_topics": generated_topics, # Store topics for categorization
2353 },
2354 )
2355 db.add(history_entry)
2356 db.commit()
2358 # Persist sources to research_resources so the assembler
2359 # can rebuild the Sources block at render time. Was
2360 # previously written INLINE into report_content via a
2361 # "## Sources" tail — the report_content refactor moves
2362 # this to structured storage matching normal research.
2363 if sources:
2364 try:
2365 from ..web.services.research_sources_service import (
2366 ResearchSourcesService,
2367 )
2369 ResearchSourcesService.save_research_sources(
2370 research_id=research_id,
2371 sources=sources,
2372 username=username,
2373 )
2374 except Exception as e:
2375 # ``password`` is a parameter of this method —
2376 # don't render a traceback that could expose it
2377 # via diagnose=True frame locals.
2378 safe_msg = redact_secrets(str(e), password)
2379 logger.warning(
2380 "Failed to persist scheduler sources for "
2381 "research {} — assembler will render no Sources "
2382 "block for this row: {}",
2383 research_id,
2384 safe_msg,
2385 )
2387 # Store the report content using storage abstraction
2388 from ..storage import get_report_storage
2390 # Use storage to save the report (report_content already retrieved above)
2391 storage = get_report_storage(session=db)
2392 storage.save_report(
2393 research_id=research_id,
2394 content=report_content,
2395 username=username,
2396 )
2398 logger.info(
2399 f"Stored research result {research_id} for subscription {subscription_id}"
2400 )
2402 except Exception as e:
2403 # ``password`` is a function parameter, so it is always in
2404 # this frame. Drop traceback + redact str(e) to avoid leaking
2405 # the SQLCipher master password.
2406 safe_msg = redact_secrets(str(e), password)
2407 logger.warning(f"Error storing research result: {safe_msg}")
2409 def _run_cleanup_with_tracking(self):
2410 """Wrapper that tracks cleanup execution."""
2412 try:
2413 cleaned_count = self._cleanup_inactive_users()
2415 logger.info(
2416 f"Cleanup successful: removed {cleaned_count} inactive users"
2417 )
2419 except Exception:
2420 logger.exception("Cleanup job failed")
2422 def _cleanup_inactive_users(self) -> int:
2423 """Remove users inactive for longer than retention period."""
2424 retention_hours = self.config.get("retention_hours", 48)
2425 cutoff = datetime.now(UTC) - timedelta(hours=retention_hours)
2427 cleaned_count = 0
2429 with self.lock:
2430 inactive_users = [
2431 user_id
2432 for user_id, session in self.user_sessions.items()
2433 if session["last_activity"] < cutoff
2434 ]
2436 for user_id in inactive_users:
2437 # Remove all scheduled jobs
2438 for job_id in self.user_sessions[user_id][
2439 "scheduled_jobs"
2440 ].copy():
2441 try:
2442 self.scheduler.remove_job(job_id)
2443 except JobLookupError:
2444 pass
2446 # Clear credentials and session data
2447 self._credential_store.clear(user_id)
2448 del self.user_sessions[user_id]
2449 cleaned_count += 1
2450 logger.info(f"Cleaned up inactive user {user_id}")
2452 return cleaned_count
2454 def _reload_config(self):
2455 """Reload configuration from settings manager."""
2456 if not hasattr(self, "settings_manager") or not self.settings_manager:
2457 return
2459 try:
2460 old_retention = self.config.get("retention_hours", 48)
2462 # Reload all settings
2463 for key in self.config:
2464 if key == "enabled":
2465 continue # Don't change enabled state while running
2467 full_key = f"news.scheduler.{key}"
2468 self.config[key] = self._get_setting(full_key, self.config[key])
2470 # Handle changes that need immediate action
2471 if old_retention != self.config["retention_hours"]:
2472 logger.info(
2473 f"Retention period changed from {old_retention} "
2474 f"to {self.config['retention_hours']} hours"
2475 )
2476 # Trigger immediate cleanup with new retention
2477 self.scheduler.add_job(
2478 self._wrap_job(self._run_cleanup_with_tracking),
2479 "date",
2480 run_date=datetime.now(UTC) + timedelta(seconds=5),
2481 id="immediate_cleanup_config_change",
2482 )
2484 # Clear settings cache to pick up any user setting changes
2485 self.invalidate_all_settings_cache()
2487 except Exception:
2488 logger.exception("Error reloading configuration")
2490 def get_status(self) -> Dict[str, Any]:
2491 """Get scheduler status information."""
2492 with self.lock:
2493 active_users = len(self.user_sessions)
2494 total_jobs = sum(
2495 len(session["scheduled_jobs"])
2496 for session in self.user_sessions.values()
2497 )
2499 # Get next run time for cleanup job
2500 next_cleanup = None
2501 if self.is_running:
2502 job = self.scheduler.get_job("cleanup_inactive_users")
2503 if job: 2503 ↛ 2506line 2503 didn't jump to line 2506 because the condition on line 2503 was always true
2504 next_cleanup = job.next_run_time
2506 return {
2507 "is_running": self.is_running,
2508 "config": self.config,
2509 "active_users": active_users,
2510 "total_scheduled_jobs": total_jobs,
2511 "next_cleanup": next_cleanup.isoformat() if next_cleanup else None,
2512 "memory_usage": self._estimate_memory_usage(),
2513 }
2515 def _estimate_memory_usage(self) -> int:
2516 """Estimate memory usage of user sessions."""
2518 # Rough estimate: username (50) + password (100) + metadata (200) per user
2519 per_user_estimate = 350
2520 return len(self.user_sessions) * per_user_estimate
2522 def get_user_sessions_summary(self) -> List[Dict[str, Any]]:
2523 """Get summary of active user sessions (without passwords)."""
2524 with self.lock:
2525 summary = []
2526 for user_id, session in self.user_sessions.items():
2527 summary.append(
2528 {
2529 "user_id": user_id,
2530 "last_activity": session["last_activity"].isoformat(),
2531 "scheduled_jobs": len(session["scheduled_jobs"]),
2532 "time_since_activity": str(
2533 datetime.now(UTC) - session["last_activity"]
2534 ),
2535 }
2536 )
2537 return summary
2540# Singleton instance getter
2541_scheduler_instance = None
2544def get_background_job_scheduler() -> BackgroundJobScheduler:
2545 """Get the singleton news scheduler instance."""
2546 global _scheduler_instance
2547 if _scheduler_instance is None:
2548 _scheduler_instance = BackgroundJobScheduler()
2549 return _scheduler_instance