Coverage for src/local_deep_research/utilities/log_utils.py: 84%
225 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-19 23:35 +0000
1"""
2Utilities for logging.
3"""
5# Needed for loguru annotations
6from __future__ import annotations
8import inspect
10# import logging - needed for InterceptHandler compatibility
11import logging
12import os
13import queue
14import sys
15import threading
16from functools import wraps
17from typing import Any, Callable
19import loguru
20from flask import g, has_app_context
21from loguru import logger
23from ..config.paths import get_logs_directory
24from ..database.models import ResearchLog
25from ..web.services.socket_service import SocketIOService
27_LOG_DIR = get_logs_directory()
28_LOG_DIR.mkdir(parents=True, exist_ok=True)
30# Thread-safe queue for database logs from background threads
31_log_queue = queue.Queue(maxsize=1000)
32_queue_processor_thread = None
33_queue_processor_lock = threading.Lock()
34_stop_queue = threading.Event()
35"""
36Default log directory to use.
37"""
39# Cap how much of a single log record's message we ship to the browser over
40# socket.io. Some diagnostic log lines (e.g. ``[FETCH] page_text``) inline
41# the full extracted page body — up to ~10 KB per call — which is useless
42# in the UI (a single massive blob fills the viewport) and inflates both
43# wire traffic and client-side state. Container-log/stderr, file, and DB
44# sinks remain unchanged, so full diagnostics are preserved for grep/DB
45# queries. The cap bounds the *prefix* preserved from the original message;
46# the wire payload is the prefix plus a short truncation indicator (~100
47# bytes), so it can exceed this value by that fixed overhead.
48FRONTEND_MESSAGE_MAX_LENGTH = 2000
50# Cap the size of messages persisted to ResearchLog. The DB column is
51# unbounded TEXT, so a long langgraph run can accumulate thousands of
52# 10 KB rows — paginated reads (PR #4037) hide the symptom but don't
53# stop the storage growth. Same prefix-plus-indicator semantics as
54# FRONTEND_MESSAGE_MAX_LENGTH; container-log/stderr/file sinks remain
55# unchanged so full diagnostics are still preserved out-of-band.
56DATABASE_MESSAGE_MAX_LENGTH = 5000
59class InterceptHandler(logging.Handler):
60 """
61 Intercepts logging messages and forwards them to Loguru's logger.
62 """
64 def emit(self, record: logging.LogRecord) -> None:
65 # Get corresponding Loguru level if it exists.
66 try:
67 level: str | int = logger.level(record.levelname).name
68 except ValueError:
69 level = record.levelno
71 # Find caller from where originated the logged message.
72 frame, depth = inspect.currentframe(), 0
73 while frame: 73 ↛ 82line 73 didn't jump to line 82 because the condition on line 73 was always true
74 filename = frame.f_code.co_filename
75 is_logging = filename == logging.__file__
76 is_frozen = "importlib" in filename and "_bootstrap" in filename
77 if depth > 0 and not (is_logging or is_frozen):
78 break
79 frame = frame.f_back
80 depth += 1
82 logger.opt(depth=depth, exception=record.exc_info).log(
83 level, record.getMessage()
84 )
87def log_for_research(
88 to_wrap: Callable[[str, ...], Any],
89) -> Callable[[str, ...], Any]:
90 """
91 Decorator for a function that's part of the research process. It expects the function to
92 take the research ID (UUID) as the first parameter, and configures all log
93 messages made during this request to include the research ID.
95 Args:
96 to_wrap: The function to wrap. Should take the research ID as the first parameter.
98 Returns:
99 The wrapped function.
101 """
103 @wraps(to_wrap)
104 def wrapped(research_id: str, *args: Any, **kwargs: Any) -> Any:
105 g.research_id = research_id
106 result = to_wrap(research_id, *args, **kwargs)
107 g.pop("research_id")
108 return result
110 return wrapped
113def _get_research_context_fallback() -> dict | None:
114 """Read the per-thread research context, if any.
116 Used as a fallback when individual log calls don't bind research_id/
117 username via ``logger.bind``. The research thread sets this once at
118 startup via ``set_search_context``, so every subsequent log call from
119 the same thread picks up research_id, username, and user_password
120 automatically — without requiring every call site to remember to bind.
121 """
122 try:
123 from .thread_context import get_search_context
125 return get_search_context()
126 except Exception:
127 return None
130def _get_research_id(record=None) -> str | None:
131 """
132 Gets the current research ID (UUID), if present.
134 Args:
135 record: Optional loguru record that might contain bound research_id
137 Returns:
138 The current research ID (UUID), or None if it does not exist.
140 """
141 # First check if research_id is bound to the log record
142 if record and "extra" in record and "research_id" in record["extra"]:
143 return record["extra"]["research_id"]
144 # Then check Flask context
145 if has_app_context():
146 gid = g.get("research_id")
147 if gid:
148 return gid
149 # Fall back to per-thread research context — research-thread logger
150 # calls without an explicit bind still get attributed correctly.
151 ctx = _get_research_context_fallback()
152 if ctx: 152 ↛ 153line 152 didn't jump to line 153 because the condition on line 152 was never true
153 return ctx.get("research_id")
154 return None
157# Counters for swallowed exceptions in the logging path. Bare except: pass
158# is required here (logging errors must not propagate or recurse), but we
159# write a stderr line on each new occurrence so silent failures aren't
160# invisible — only active when LDR_APP_DEBUG=true so production stderr
161# stays clean.
162_silent_exc_counts: dict[str, int] = {}
165def _report_silent_exception(
166 where: str,
167 exc_type_name: str,
168 username: str | None = None,
169 research_id: str | None = None,
170 level: str | None = None,
171) -> None:
172 """Surface a swallowed logging-path exception to stderr.
174 Bypasses ``logger`` to avoid recursing back through ``database_sink``.
175 Rate-limited to first occurrence + every 100th repeat for the same
176 ``where`` key, so a persistent failure mode doesn't flood the console.
178 Note: takes the exception's TYPE NAME as a plain string (not the
179 exception object). The caller does ``type(exc).__name__`` and passes
180 the result. This is deliberate — CodeQL's taint analyzer treats any
181 function frame holding a ``BaseException`` captured from a password-
182 bearing call site as tainted, and flags every stderr write inside
183 that frame. Receiving only a type-name string severs the flow at
184 the boundary.
185 """
186 if os.environ.get("LDR_APP_DEBUG", "").lower() not in ("1", "true", "yes"): 186 ↛ 188line 186 didn't jump to line 188 because the condition on line 186 was always true
187 return
188 n = _silent_exc_counts.get(where, 0) + 1
189 _silent_exc_counts[where] = n
190 if n != 1 and n % 100 != 0:
191 return
192 parts = []
193 if username is not None:
194 parts.append(f"username={username!r}")
195 if research_id is not None:
196 parts.append(f"research_id={research_id!r}")
197 if level is not None:
198 parts.append(f"level={level!r}")
199 ctx = " ".join(parts)
200 # CodeQL's py/clear-text-logging-sensitive-data may flag this stderr
201 # write because the function frame is reachable from
202 # _write_log_to_database which holds user_password locally. The data
203 # actually written is only plain strings — `where` (literal),
204 # `exc_type_name` (`type(exc).__name__`), and `username/research_id/level`
205 # repr'd from the queue entry. No password value ever reaches the
206 # formatter; the helper signature deliberately accepts only typed
207 # primitives. If CodeQL flags this line, dismiss the alert as a
208 # false positive in the Security tab with that justification.
209 sys.stderr.write(
210 f"[log-utils] {where} swallowed (count={n}): "
211 f"{exc_type_name}{(' ' + ctx) if ctx else ''}\n"
212 )
213 sys.stderr.flush()
216def _process_log_queue():
217 """
218 Process logs from the queue in a dedicated daemon thread.
220 Safe to run off the main thread: ``_write_log_to_database`` uses
221 ``get_user_db_session`` which yields a thread-local SQLAlchemy session,
222 and the underlying SQLite engines are opened with
223 ``check_same_thread=False``.
224 """
225 while not _stop_queue.is_set():
226 try:
227 # Wait for logs with timeout to check stop flag
228 log_entry = _log_queue.get(timeout=0.1)
230 # Skip if no entry
231 if log_entry is None:
232 continue
234 # Write to database if we have app context
235 if has_app_context():
236 _write_log_to_database(log_entry)
237 else:
238 # If no app context, put it back in queue for later
239 try:
240 _log_queue.put_nowait(log_entry)
241 except queue.Full:
242 pass # Drop log if queue is full
244 except queue.Empty:
245 continue
246 except Exception as exc:
247 # noqa: silent-exception — must not let logging errors crash the log processor thread.
248 # Wrap the report itself: if stderr is closed (broken pipe etc.)
249 # an IOError from inside an except handler propagates and would
250 # kill the daemon thread, silently breaking all subsequent log
251 # persistence for the rest of the process lifetime.
252 try:
253 _report_silent_exception(
254 "process_log_queue", type(exc).__name__
255 )
256 except Exception:
257 pass # noqa: silent-exception — broken stderr must not kill the daemon
260def _write_log_to_database(log_entry: dict) -> None:
261 """
262 Write a log entry to the database. Should only be called from main thread.
263 """
264 from ..database.session_context import get_user_db_session
266 try:
267 username = log_entry.get("username")
268 # Captured in the emitting thread (database_sink) from
269 # ContextVar storage; the queue daemon thread can't read that itself.
270 pw = log_entry.get("user_password") # gitleaks:allow
272 with get_user_db_session(
273 username, password=pw
274 ) as db_session: # gitleaks:allow
275 if db_session: 275 ↛ exitline 275 didn't jump to the function exit
276 db_log = ResearchLog(
277 timestamp=log_entry["timestamp"],
278 message=log_entry["message"],
279 module=log_entry["module"],
280 function=log_entry["function"],
281 line_no=log_entry["line_no"],
282 level=log_entry["level"],
283 research_id=log_entry["research_id"],
284 )
285 db_session.add(db_log)
286 db_session.commit()
287 except Exception as exc:
288 # noqa: silent-exception — DB errors in the logging path must not propagate or recurse.
289 # Wrap the report itself so a broken-stderr IOError can't escape and
290 # be re-caught by an outer logging-aware handler somewhere upstream.
291 try:
292 _report_silent_exception(
293 "write_log_to_database",
294 type(exc).__name__,
295 username=log_entry.get("username"),
296 research_id=log_entry.get("research_id"),
297 level=log_entry.get("level"),
298 )
299 except Exception:
300 pass # noqa: silent-exception — broken stderr must not bubble out of logging path
303def database_sink(message: loguru.Message) -> None:
304 """
305 Sink that saves messages to the database.
306 Queues logs from background threads for later processing.
308 Args:
309 message: The log message to save.
311 """
312 record = message.record
313 research_id = _get_research_id(record)
315 # Resolve username + password. The queue daemon thread can't read the
316 # research thread's ContextVar storage and has no Flask request
317 # context, so we capture both here in the emitting thread.
318 #
319 # Source priority:
320 # 1. logger.bind(...) extras on the record itself
321 # 2. per-thread research context (set once when the research thread
322 # starts, so every subsequent log call inherits it without
323 # requiring an explicit bind)
324 # 3. Flask request session (for request-handler threads that
325 # tagged research_id via the @log_for_research decorator but
326 # didn't bind username — common for /api/research/<id>/* routes)
327 username = record.get("extra", {}).get("username")
328 user_password = None
329 ctx = _get_research_context_fallback()
330 if ctx:
331 if not username: 331 ↛ 333line 331 didn't jump to line 333 because the condition on line 331 was always true
332 username = ctx.get("username")
333 user_password = ctx.get("user_password")
334 # Only consult Flask request state when the log already has a
335 # research_id. ResearchLog is research-scoped by design — auth and
336 # other system DEBUG logs (research_id=None) don't belong there. If
337 # we attached a username to them via flask_session, they'd just churn
338 # through the queue and fail at the daemon (where the encrypted DB
339 # may not even be open yet for that user — e.g. right after a server
340 # restart with a still-valid session cookie).
341 if research_id is not None and has_app_context():
342 try:
343 from flask import session as flask_session, has_request_context
345 if not username and has_request_context(): 345 ↛ 346line 345 didn't jump to line 346 because the condition on line 345 was never true
346 username = flask_session.get("username")
347 # Password is set on g.user_password by the request middleware
348 # after authentication. The daemon thread can't read this, so
349 # capture it here in the request thread.
350 if not user_password and hasattr(g, "user_password"): 350 ↛ 358line 350 didn't jump to line 358 because the condition on line 350 was always true
351 user_password = g.user_password
352 except Exception:
353 pass # noqa: silent-exception — must not fail logging on session lookup
355 # Skip persistence for system logs that have no research context.
356 # These can't be written to any per-user encrypted DB and would just
357 # churn through the queue + daemon for no useful end state.
358 if research_id is None and username is None:
359 return
361 # Create log entry dict
362 log_entry = {
363 "timestamp": record["time"],
364 "message": _truncate_for_database(record["message"]),
365 "module": record["name"],
366 "function": record["function"],
367 "line_no": int(record["line"]),
368 "level": record["level"].name,
369 "research_id": research_id,
370 "username": username,
371 "user_password": user_password,
372 }
374 # Check if we're in a background thread
375 # Note: Socket.IO handlers run in separate threads even with app context
376 if not has_app_context() or threading.current_thread().name != "MainThread":
377 # Queue the log for later processing
378 try:
379 _log_queue.put_nowait(log_entry)
380 except queue.Full:
381 # Drop log if queue is full to avoid blocking
382 pass
383 else:
384 # We're in the main thread with app context - write directly
385 _write_log_to_database(log_entry)
388def _truncate_for_database(message: str) -> str:
389 """Bound the persisted size of a log message.
391 ``DATABASE_MESSAGE_MAX_LENGTH`` is the preserved-prefix length;
392 the stored string is at most that plus a ~100-char suffix that
393 reports the original length so debug context is not lost. Full
394 messages remain available in container-log/stderr/file sinks.
395 """
396 if len(message) <= DATABASE_MESSAGE_MAX_LENGTH:
397 return message
398 suffix = (
399 f"… (truncated; full message in server logs; "
400 f"original length: {len(message)} chars)"
401 )
402 return message[:DATABASE_MESSAGE_MAX_LENGTH] + suffix
405def _truncate_for_frontend(message: str) -> str:
406 """Bound the wire size of an outbound log message.
408 ``FRONTEND_MESSAGE_MAX_LENGTH`` caps the *preserved prefix* of the
409 original message. When truncation kicks in, a short indicator is
410 appended that names the original length and points the user at the
411 server-side logs for the full text, so the returned string is
412 ``FRONTEND_MESSAGE_MAX_LENGTH`` plus the fixed indicator overhead
413 (~100 bytes). Verbose diagnostic logs (e.g. ``[FETCH] page_text``
414 which inlines the full extracted page body) are useless in the UI
415 when displayed in full and inflate socket payloads + client-side
416 memory; container-log/stderr, file, and DB sinks remain unchanged.
417 """
418 if len(message) <= FRONTEND_MESSAGE_MAX_LENGTH:
419 return message
420 suffix = (
421 f"… (truncated; full message in server logs; "
422 f"original length: {len(message)} chars)"
423 )
424 return message[:FRONTEND_MESSAGE_MAX_LENGTH] + suffix
427def frontend_progress_sink(message: loguru.Message) -> None:
428 """
429 Sink that sends messages to the frontend.
431 Args:
432 message: The log message to send.
434 """
435 record = message.record
436 research_id = _get_research_id(record)
437 if research_id is None:
438 # If we don't have a research ID, don't send anything.
439 # Can't use logger here as it causes deadlock
440 return
442 # Defence in depth (R4-09): never forward policy-audit log lines
443 # to WebSocket subscribers. They carry engine names + reason codes
444 # which could leak the active scope to a cross-origin observer under
445 # CORS=*. Today policy_audit logs don't bind research_id so the
446 # research_id guard above already skips them; this filter is the
447 # explicit guarantee in case a future call site binds both.
448 if record.get("extra", {}).get("policy_audit"):
449 return
451 frontend_log = {
452 "log_entry": {
453 "message": _truncate_for_frontend(record["message"]),
454 "type": record["level"].name, # Keep original case
455 "time": record["time"].isoformat(),
456 },
457 }
458 SocketIOService().emit_to_subscribers(
459 "progress", research_id, frontend_log, enable_logging=False
460 )
463def flush_log_queue():
464 """
465 Drain all pending logs from the queue to the database.
467 Used at process exit (see ``flush_logs_on_exit`` in ``web/app.py``) to
468 drain whatever the background daemon did not get to before it was
469 stopped. The request path no longer calls this — the
470 ``start_log_queue_processor`` daemon handles steady-state drainage so
471 requests never block on DB writes.
472 """
473 flushed = 0
474 while not _log_queue.empty():
475 try:
476 log_entry = _log_queue.get_nowait()
477 _write_log_to_database(log_entry)
478 flushed += 1
479 except queue.Empty:
480 break
481 except Exception:
482 pass # noqa: silent-exception — DB errors during log flush must not propagate
484 if flushed > 0:
485 logger.debug(f"Flushed {flushed} queued log entries to database")
488def start_log_queue_processor(app) -> threading.Thread:
489 """Start the background daemon that drains the log queue into the DB.
491 Runs ``_process_log_queue`` inside an application context so writes
492 have a Flask context, and so the daemon can work independently of
493 any in-flight request. Idempotent: calling twice is a no-op.
495 Args:
496 app: The Flask app whose context the daemon should push.
498 Returns:
499 The daemon thread (running).
500 """
501 global _queue_processor_thread
502 with _queue_processor_lock:
503 if (
504 _queue_processor_thread is not None
505 and _queue_processor_thread.is_alive()
506 ):
507 return _queue_processor_thread
509 _stop_queue.clear()
511 def _run():
512 # Push an app context for the lifetime of the daemon so
513 # the queue processor can call into DB code that requires
514 # Flask g.
515 with app.app_context():
516 _process_log_queue()
518 _queue_processor_thread = threading.Thread(
519 target=_run,
520 name="log-queue-processor",
521 daemon=True,
522 )
523 _queue_processor_thread.start()
524 thread = _queue_processor_thread
525 logger.info("Log queue processor daemon started")
526 return thread
529def stop_log_queue_processor(timeout: float = 2.0) -> None:
530 """Signal the log queue processor to stop and wait briefly for it."""
531 global _queue_processor_thread
532 _stop_queue.set()
533 with _queue_processor_lock:
534 thread = _queue_processor_thread
535 if thread is not None: 535 ↛ exitline 535 didn't return from function 'stop_log_queue_processor' because the condition on line 535 was always true
536 thread.join(timeout=timeout)
537 # Only clear the reference if the thread actually exited. If join
538 # timed out the daemon is still running, and clearing the ref would
539 # let a subsequent start_log_queue_processor() spawn a second
540 # daemon that drains the same queue concurrently. Re-check identity
541 # under the lock so we don't accidentally null out a fresh thread
542 # that another start spawned in the meantime.
543 if not thread.is_alive():
544 with _queue_processor_lock:
545 if _queue_processor_thread is thread: 545 ↛ exitline 545 didn't jump to the function exit
546 _queue_processor_thread = None
549def config_logger(name: str, debug: bool = False) -> None:
550 """
551 Configures the default logger.
553 Args:
554 name: The name to use for the log file.
555 debug: Whether to enable unsafe debug logging.
557 """
558 from ..security.log_sanitizer import strip_control_chars
560 def _sanitize_record(record):
561 record["message"] = strip_control_chars(record["message"])
563 logger.configure(patcher=_sanitize_record)
565 logger.enable("local_deep_research")
566 logger.remove()
568 # Log to console (stderr) and database
569 stderr_level = "DEBUG" if debug else "INFO"
571 # loguru's diagnose=True renders repr() of every local variable in every
572 # traceback frame on exceptions. Under LDR_APP_DEBUG that would dump
573 # credentials living in frame locals (api_key, SQLCipher password,
574 # Authorization headers) into every sink. Gate diagnose behind a separate
575 # explicit opt-in so enabling LDR_APP_DEBUG for general debug output does
576 # not also enable localvar dumps. Default OFF even when debug is on.
577 #
578 # security/secure_logging.py gates provider/engine exception tracebacks
579 # on the same two flags (its is_diagnose_mode()). Deliberate divergence:
580 # the wrapper reads LDR_APP_DEBUG from the environment only, while the
581 # ``debug`` argument here may also come from the app.debug DB setting /
582 # legacy server_config.json — so DB-enabled debug affects this sink's
583 # diagnose but never the wrapper's traceback gate. Kept inline (not
584 # env_truthy) because log_utils must not import security at module
585 # level: security/__init__.py runs install_audit_hook() on import.
586 diagnose = debug and os.environ.get(
587 "LDR_LOGURU_DIAGNOSE", ""
588 ).strip().lower() in ("1", "true", "yes")
590 # ``diagnose`` renders the repr() of every frame-local — which can
591 # include the SQLCipher master password and other credentials — into
592 # the rendered exception block. Allow it ONLY on the ephemeral, local
593 # stderr sink the operator explicitly opted into. The database sink
594 # PERSISTS into the user's own encrypted DB and the frontend sink
595 # SHIPS to the browser, so they must NEVER render frame locals, even
596 # under LDR_LOGURU_DIAGNOSE. This single chokepoint protects every
597 # credential-bearing exception handler app-wide against the
598 # frame-locals leak, independent of per-site logging discipline
599 # (#4182).
600 #
601 # enqueue=True on stderr: loguru emits to an in-memory queue and a
602 # single background thread does the actual stderr write, so a log call
603 # never blocks on stderr I/O while holding the handler's lock. Without
604 # it, under the werkzeug threading dev server every request thread logs
605 # synchronously, and when stderr back-pressures (e.g. a slow/full
606 # `docker logs` pipe in CI) the lock-holder blocks mid-write and ALL
607 # logging threads — i.e. all request threads — pile up behind the lock,
608 # freezing the whole request pipeline for ~60s (#4431). Captured
609 # forensically: 3/5 server threads parked in loguru's _protected_lock
610 # under load. The database/progress sinks keep their own
611 # emitting-thread context capture and are left synchronous.
612 logger.add(sys.stderr, level=stderr_level, diagnose=diagnose, enqueue=True)
613 logger.add(database_sink, level="DEBUG", diagnose=False)
614 logger.add(frontend_progress_sink, diagnose=False)
616 if debug:
617 logger.warning(
618 "DEBUG logging is enabled (LDR_APP_DEBUG=true). "
619 "Logs may contain sensitive data (queries, answers, API responses). "
620 "Do NOT use in production."
621 )
623 if diagnose:
624 logger.warning(
625 "LDR_LOGURU_DIAGNOSE is enabled: exception tracebacks will include "
626 "local variable values, which may contain credentials (API keys, "
627 "passwords, tokens). Do NOT use in production."
628 )
630 # Optionally log to file if enabled (disabled by default for security)
631 # Check environment variable first, then database setting
632 enable_file_logging = (
633 os.environ.get("LDR_ENABLE_FILE_LOGGING", "").lower() == "true"
634 )
636 # File logging is controlled only by environment variable for simplicity
637 # Database settings are not available at logger initialization time
639 if enable_file_logging:
640 log_file = _LOG_DIR / f"{name}.log"
641 logger.add(
642 log_file,
643 level="DEBUG",
644 rotation="10 MB",
645 retention="7 days",
646 compression="zip",
647 # diagnose=False: the file sink is persistent and unencrypted,
648 # so — like the DB and frontend sinks — it must never render
649 # frame-local credentials, even under LDR_LOGURU_DIAGNOSE.
650 # Frame-local dumps go only to the ephemeral stderr sink, which
651 # is what the policy comment above describes (#4182).
652 diagnose=False,
653 )
654 logger.warning(
655 f"File logging enabled - logs will be written to {log_file}. "
656 "WARNING: Log files are unencrypted and may contain sensitive data!"
657 )
659 # Add a special log level for milestones.
660 try:
661 logger.level("MILESTONE", no=26, color="<magenta><bold>")
662 except ValueError:
663 # Level already exists, that's fine
664 pass