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