Coverage for src/local_deep_research/utilities/log_utils.py: 85%
281 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-06 15:42 +0000
1"""
2Utilities for logging.
3"""
5# Needed for loguru annotations
6from __future__ import annotations
8import asyncio
9import inspect
11# import logging - needed for InterceptHandler compatibility
12import logging
13import os
14import queue
15import sys
16import contextvars
17import threading
18from functools import wraps
19from typing import Any, Callable
21import loguru
22from loguru import logger
24from ..config.paths import get_logs_directory
25from ..database.models import ResearchLog
26from ..web.services.socketio_asgi import emit_to_subscribers
29# Per-task context for research_id, replacing Flask's g.research_id.
30# ContextVar works in both sync and async code and is correctly isolated
31# per-task in asyncio (unlike threading.local in async code).
32_research_id_var: contextvars.ContextVar[str | None] = contextvars.ContextVar(
33 "research_id", default=None
34)
36_LOG_DIR = get_logs_directory()
37# Lazy import: log_utils is imported very early by many bootstrap paths, so
38# importing security at module top level risks an import cycle.
39from ..security.directory_creation import create_directory # noqa: E402
41create_directory(_LOG_DIR, context="logs directory")
44def _register_milestone_level() -> None:
45 """Register the custom MILESTONE level.
47 Called at import rather than only from ``config_logger()``. Loguru levels
48 are process-global, and ``research_service``'s ``progress_callback`` opens
49 with ``bound_logger.log("MILESTONE", ...)`` — so in any process that never
50 called ``config_logger()`` (i.e. every pytest run; it is invoked only from
51 ``web/app.py``'s ``ldr-web`` entry point) that first call raises
52 ``ValueError: Level 'MILESTONE' does not exist``.
54 The consequence was not a visible failure. The exception surfaced at the
55 worker's first progress checkpoint, was re-raised inside the error handler
56 and swallowed by a broad ``except``, so the research thread died before
57 writing any terminal status and no assertion failed. Whether a given test
58 process had the level at all depended on whether an unrelated module that
59 registers it (two test modules do so at import time) had been collected
60 first — i.e. research-worker tests were order-dependent by construction.
61 """
62 try:
63 logger.level("MILESTONE", no=26, color="<magenta><bold>")
64 except ValueError:
65 # Already registered — levels are process-global and permanent.
66 pass
69_register_milestone_level()
71# Thread-safe queue for database logs from background threads
72_log_queue = queue.Queue(maxsize=1000)
74# Thread-local re-entry guard for `database_sink`. If a `logger.*` call
75# fires while we're inside the sink (e.g. loguru's own error handler),
76# we must not re-enter or the process deadlocks/recurses.
77_sink_state = threading.local()
78_queue_processor_thread = None
79_queue_processor_lock = threading.Lock()
80_stop_queue = threading.Event()
81"""
82Default log directory to use.
83"""
85# Cap how much of a single log record's message we ship to the browser over
86# socket.io. Some diagnostic log lines (e.g. ``[FETCH] page_text``) inline
87# the full extracted page body — up to ~10 KB per call — which is useless
88# in the UI (a single massive blob fills the viewport) and inflates both
89# wire traffic and client-side state. Container-log/stderr, file, and DB
90# sinks remain unchanged, so full diagnostics are preserved for grep/DB
91# queries. The cap bounds the *prefix* preserved from the original message;
92# the wire payload is the prefix plus a short truncation indicator (~100
93# bytes), so it can exceed this value by that fixed overhead.
94FRONTEND_MESSAGE_MAX_LENGTH = 2000
96# Cap the size of messages persisted to ResearchLog. The DB column is
97# unbounded TEXT, so a long langgraph run can accumulate thousands of
98# 10 KB rows — paginated reads (PR #4037) hide the symptom but don't
99# stop the storage growth. Same prefix-plus-indicator semantics as
100# FRONTEND_MESSAGE_MAX_LENGTH; container-log/stderr/file sinks remain
101# unchanged so full diagnostics are still preserved out-of-band.
102DATABASE_MESSAGE_MAX_LENGTH = 5000
105class InterceptHandler(logging.Handler):
106 """
107 Intercepts logging messages and forwards them to Loguru's logger.
108 """
110 def emit(self, record: logging.LogRecord) -> None:
111 # Get corresponding Loguru level if it exists.
112 try:
113 level: str | int = logger.level(record.levelname).name
114 except ValueError:
115 level = record.levelno
117 # Find caller from where originated the logged message.
118 frame, depth = inspect.currentframe(), 0
119 while frame: 119 ↛ 128line 119 didn't jump to line 128 because the condition on line 119 was always true
120 filename = frame.f_code.co_filename
121 is_logging = filename == logging.__file__
122 is_frozen = "importlib" in filename and "_bootstrap" in filename
123 if depth > 0 and not (is_logging or is_frozen):
124 break
125 frame = frame.f_back
126 depth += 1
128 logger.opt(depth=depth, exception=record.exc_info).log(
129 level, record.getMessage()
130 )
133def log_for_research(
134 to_wrap: Callable[[str, ...], Any],
135) -> Callable[[str, ...], Any]:
136 """
137 Decorator for a function that's part of the research process. It expects the function to
138 take the research ID (UUID) as the first parameter, and configures all log
139 messages made during this request to include the research ID.
141 Args:
142 to_wrap: The function to wrap. Should take the research ID as the first parameter.
144 Returns:
145 The wrapped function.
147 """
149 @wraps(to_wrap)
150 def wrapped(research_id: str, *args: Any, **kwargs: Any) -> Any:
151 cv_handle = _research_id_var.set(research_id)
152 try:
153 return to_wrap(research_id, *args, **kwargs)
154 finally:
155 _research_id_var.reset(cv_handle)
157 return wrapped
160def _get_research_context_fallback() -> dict | None:
161 """Read the per-thread research context, if any.
163 Used as a fallback when individual log calls don't bind research_id/
164 username via ``logger.bind``. The research thread sets this once at
165 startup via ``set_search_context``, so every subsequent log call from
166 the same thread picks up research_id, username, and user_password
167 automatically — without requiring every call site to remember to bind.
168 """
169 try:
170 from .thread_context import get_search_context
172 return get_search_context()
173 except Exception:
174 return None
177def _get_request_username() -> str | None:
178 """Read the request-scoped username, if any.
180 The FastAPI successor to main's ``flask.session.get("username")``
181 fallback in the log sinks. Reads the contextvar ``DatabaseMiddleware``
182 populates per request (and that background workers push explicitly via
183 ``request_user``), so a log call that binds ``research_id`` without a
184 ``username`` -- as report_assembly_service does from a real request
185 thread -- still resolves an owner instead of being dropped.
187 Imported lazily and never allowed to raise: this runs inside a loguru
188 sink, where an exception would break logging itself.
189 """
190 try:
191 from .request_context import get_current_username
193 return get_current_username()
194 except Exception:
195 return None
198def _get_research_id(record=None) -> str | None:
199 """
200 Gets the current research ID (UUID), if present.
202 Args:
203 record: Optional loguru record that might contain bound research_id
205 Returns:
206 The current research ID (UUID), or None if it does not exist.
207 """
208 # First check if research_id is bound to the log record
209 if record and "extra" in record and "research_id" in record["extra"]:
210 return record["extra"]["research_id"]
211 # Then check the contextvar (set by @log_for_research on the running
212 # task/thread)
213 rid = _research_id_var.get()
214 if rid:
215 return rid
216 # Fall back to per-thread research context — research-thread logger
217 # calls without an explicit bind still get attributed correctly.
218 ctx = _get_research_context_fallback()
219 if ctx:
220 return ctx.get("research_id")
221 return None
224# Counters for swallowed exceptions in the logging path. Bare except: pass
225# is required here (logging errors must not propagate or recurse), but we
226# write a stderr line on each new occurrence so silent failures aren't
227# invisible — only active when LDR_APP_DEBUG=true so production stderr
228# stays clean.
229_silent_exc_counts: dict[str, int] = {}
232def _report_silent_exception(
233 where: str,
234 exc_type_name: str,
235 username: str | None = None,
236 research_id: str | None = None,
237 level: str | None = None,
238) -> None:
239 """Surface a swallowed logging-path exception to stderr.
241 Bypasses ``logger`` to avoid recursing back through ``database_sink``.
242 Rate-limited to first occurrence + every 100th repeat for the same
243 ``where`` key, so a persistent failure mode doesn't flood the console.
245 Note: takes the exception's TYPE NAME as a plain string (not the
246 exception object). The caller does ``type(exc).__name__`` and passes
247 the result. This is deliberate — CodeQL's taint analyzer treats any
248 function frame holding a ``BaseException`` captured from a password-
249 bearing call site as tainted, and flags every stderr write inside
250 that frame. Receiving only a type-name string severs the flow at
251 the boundary.
252 """
253 if os.environ.get("LDR_APP_DEBUG", "").lower() not in ("1", "true", "yes"): 253 ↛ 255line 253 didn't jump to line 255 because the condition on line 253 was always true
254 return
255 n = _silent_exc_counts.get(where, 0) + 1
256 _silent_exc_counts[where] = n
257 if n != 1 and n % 100 != 0:
258 return
259 parts = []
260 if username is not None:
261 parts.append(f"username={username!r}")
262 if research_id is not None:
263 parts.append(f"research_id={research_id!r}")
264 if level is not None:
265 parts.append(f"level={level!r}")
266 ctx = " ".join(parts)
267 # CodeQL's py/clear-text-logging-sensitive-data may flag this stderr
268 # write because the function frame is reachable from the logging path.
269 # (_write_log_to_database no longer holds any credential at all since
270 # #5538 -- it binds to the already-open engine.) The data
271 # actually written is only plain strings — `where` (literal),
272 # `exc_type_name` (`type(exc).__name__`), and `username/research_id/level`
273 # repr'd from the queue entry. No password value ever reaches the
274 # formatter; the helper signature deliberately accepts only typed
275 # primitives. If CodeQL flags this line, dismiss the alert as a
276 # false positive in the Security tab with that justification.
277 sys.stderr.write(
278 f"[log-utils] {where} swallowed (count={n}): "
279 f"{exc_type_name}{(' ' + ctx) if ctx else ''}\n"
280 )
281 sys.stderr.flush()
284def _clear_daemon_thread_credentials() -> None:
285 """Drop any DB session / cached credential held by the current thread.
287 Invoked at the end of every log-queue drain iteration. The queue
288 daemon is a long-lived ``daemon=True`` thread whose credentials the
289 dead-thread sweeper (``cleanup_dead_threads``) can never reclaim,
290 because the thread stays alive for the whole process. Clearing here
291 guarantees the ``log-queue-processor`` thread never retains a
292 plaintext credential in ``thread_local_session``'s
293 ``_thread_credentials``.
295 Best-effort: must never raise, or a single cleanup hiccup would crash
296 the daemon and silently stop all subsequent log persistence.
297 """
298 try:
299 from ..database.thread_local_session import cleanup_current_thread
301 cleanup_current_thread()
302 except Exception:
303 pass # noqa: silent-exception — cleanup must not kill the log daemon
306def _process_log_queue():
307 """Process logs from the queue in a dedicated background thread.
309 Under FastAPI there's no Flask app context to gate on — we just
310 drain the queue and write entries directly. Main gated this on
311 ``has_app_context()`` and re-queued otherwise; with no such context
312 here the write is unconditional, which is safe because
313 ``_write_log_to_database`` now refuses to open a closed database.
315 Safe to run off the main thread: the write binds a session to the
316 user's already-open engine, and the underlying SQLite engines are
317 opened with ``check_same_thread=False``. It never reopens a closed
318 database, so a post-logout backlog can't resurrect a user's
319 connection from this thread (#5538). Errors are swallowed rather
320 than crashing the processor.
321 """
322 while not _stop_queue.is_set():
323 try:
324 log_entry = _log_queue.get(timeout=0.1)
325 if log_entry is None:
326 continue
327 try:
328 _write_log_to_database(log_entry)
329 finally:
330 # Belt-and-braces credential hygiene (#5538). This daemon is
331 # a long-lived daemon=True thread, so the dead-thread
332 # credential sweeper (cleanup_dead_threads) never reclaims
333 # it. Clear any per-thread DB session / cached credential at
334 # the end of every drain iteration so the
335 # log-queue-processor thread can never accumulate a
336 # plaintext credential in _thread_credentials.
337 _clear_daemon_thread_credentials()
338 except queue.Empty:
339 continue
340 except Exception as exc:
341 # noqa: silent-exception — must not let logging errors crash the log processor thread.
342 # Wrap the report itself: if stderr is closed (broken pipe etc.)
343 # an IOError from inside an except handler propagates and would
344 # kill the daemon thread, silently breaking all subsequent log
345 # persistence for the rest of the process lifetime.
346 try:
347 _report_silent_exception(
348 "process_log_queue", type(exc).__name__
349 )
350 except Exception:
351 pass # noqa: silent-exception — broken stderr must not kill the daemon
354def _write_log_to_database(log_entry: dict) -> None:
355 """Persist one queued log entry into the user's already-open database.
357 The queue drain must NEVER reopen a closed database. A log entry that
358 was queued while a user was active can be drained by the background
359 daemon *after* that user has logged out; reopening their encrypted DB
360 here would silently re-add the connection (``is_user_connected`` would
361 flip back to True) and re-cache their credentials — resurrecting
362 session state with no user action and no attacker involved.
364 So we gate strictly on an already-open connection
365 (``db_manager.is_user_connected``) and bind to the EXISTING engine via
366 ``db_manager.get_session`` — which never opens a new database and needs
367 no password. Entries for a user with no live connection (e.g. a
368 post-logout backlog) are dropped; the full message is still available
369 in the stderr/file sinks.
371 In-flight logging is preserved: while a user is active — including a
372 running research job, which keeps their DB open — the connection stays
373 live, so their logs continue to be written normally.
374 """
375 from ..database.encrypted_db import db_manager
377 username = log_entry.get("username")
378 if not username:
379 return
381 # Gate on an already-open connection. Do NOT lazily reopen the DB:
382 # after logout the connection is gone, and reopening it here would
383 # resurrect the user's connected state from a stale queued backlog.
384 if not db_manager.is_user_connected(username):
385 return
387 db_session = None
388 try:
389 # Binds a session to the ALREADY-open engine. Returns None (never
390 # reopens) if the connection was closed in the small race between
391 # the is_user_connected check above and this call. We can't use
392 # get_user_db_session here: it would resolve+cache a password and
393 # lazily reopen the DB, which is exactly the post-logout
394 # resurrection this fix removes. The session is explicitly closed
395 # in the finally block below, so the QueuePool connection is
396 # returned and no FD leaks.
397 db_session = db_manager.get_session(username) # noqa: raw-session
398 if db_session is not None: 398 ↛ 429line 398 didn't jump to line 429 because the condition on line 398 was always true
399 db_log = ResearchLog(
400 timestamp=log_entry["timestamp"],
401 message=log_entry["message"],
402 module=log_entry["module"],
403 function=log_entry["function"],
404 line_no=log_entry["line_no"],
405 level=log_entry["level"],
406 research_id=log_entry["research_id"],
407 )
408 db_session.add(db_log)
409 db_session.commit()
410 except Exception as exc:
411 # noqa: silent-exception — DB errors in the logging path must not propagate or recurse.
412 # Wrap the report itself so a broken-stderr IOError can't escape and
413 # be re-caught by an outer logging-aware handler somewhere upstream.
414 try:
415 _report_silent_exception(
416 "write_log_to_database",
417 type(exc).__name__,
418 username=log_entry.get("username"),
419 research_id=log_entry.get("research_id"),
420 level=log_entry.get("level"),
421 )
422 except Exception:
423 pass # noqa: silent-exception — broken stderr must not bubble out of logging path
424 finally:
425 # The session is freshly created per drain, bound to the shared
426 # per-user engine — close it so its pooled connection is returned
427 # (close() also rolls back any pending transaction from a failed
428 # commit). Never let a close error escape the logging path.
429 if db_session is not None:
430 try:
431 db_session.close()
432 except Exception:
433 pass # noqa: silent-exception — session close failure must not propagate
436def database_sink(message: loguru.Message) -> None:
437 """
438 Sink that saves messages to the database.
439 Queues logs from background threads for later processing.
441 Args:
442 message: The log message to save.
444 """
445 # Thread-local re-entry guard. Any code path inside this sink (or
446 # `_write_log_to_database`) that happens to call a logger — including
447 # loguru's own sink-error handler when `catch=True` is set — would
448 # re-invoke this function and recurse. The `try/except Exception: pass`
449 # in `_write_log_to_database` catches most cases, but a nested sink
450 # error handler can still land here. Fail closed for the nested call.
451 if getattr(_sink_state, "in_sink", False): 451 ↛ 452line 451 didn't jump to line 452 because the condition on line 451 was never true
452 return
453 _sink_state.in_sink = True
454 try:
455 record = message.record
456 research_id = _get_research_id(record)
458 # Resolve the username the log belongs to. The queue daemon thread
459 # can't read the research thread's ContextVar storage, so we capture
460 # it here in the emitting thread.
461 #
462 # We deliberately do NOT capture the user's password. The queue entry
463 # must never carry a plaintext credential: the background drain would
464 # otherwise use it to reopen the user's encrypted DB after logout,
465 # reinstating their connected state (#5538). The drain writes only
466 # when the DB is already open -- see _write_log_to_database.
467 #
468 # Source priority:
469 # 1. logger.bind(...) extras on the record itself
470 # 2. per-thread research context (set once when the research thread
471 # starts, so every subsequent log call inherits it without
472 # requiring an explicit bind)
473 username = record.get("extra", {}).get("username")
474 ctx = _get_research_context_fallback()
475 if ctx and not username: 475 ↛ 476line 475 didn't jump to line 476 because the condition on line 475 was never true
476 username = ctx.get("username")
477 if research_id is not None and not username:
478 # Third source, ported from main: it resolved the request user
479 # via `flask.session.get("username")` when a research_id was
480 # bound but no username was. That fallback is not dead code --
481 # report_assembly_service binds research_id alone from a real
482 # request thread, and without this the entry resolves to
483 # username=None and is silently dropped by the `if not username`
484 # guard downstream. get_current_username() is the direct
485 # successor (the contextvar DatabaseMiddleware populates).
486 #
487 # The `research_id is not None` condition is main's, and it is
488 # load-bearing for data retention, not just noise control.
489 # DatabaseMiddleware populates that contextvar for EVERY
490 # authenticated request, so without the gate any INFO record
491 # emitted during any request resolves an owner, clears the skip
492 # guard below, and is persisted to that user's app_logs with
493 # research_id = NULL. Those rows are unreachable by the only
494 # mechanism that ever removes app_logs -- the ON DELETE CASCADE
495 # from research_history -- so neither "delete this research" nor
496 # "clear all history" can touch them, and there is no retention
497 # job. They would accumulate permanently with no way to delete
498 # them. main's comment put it as: "ResearchLog is research-scoped
499 # by design -- auth and other system DEBUG logs (research_id=None)
500 # don't belong there."
501 username = _get_request_username()
503 # Skip persistence for system logs that have no research context.
504 # These can't be written to any per-user encrypted DB and would just
505 # churn through the queue + daemon for no useful end state.
506 if research_id is None and username is None:
507 return
509 # Create log entry dict
510 log_entry = {
511 "timestamp": record["time"],
512 # Prefix the message with exception type/value when the record
513 # carries one (logger.exception / logger.opt(exception=...)).
514 # Keeps the full traceback off the encrypted-DB row (diagnose=False
515 # on this sink; password-redaction policy #4182) while still
516 # leaving the database log useful for triage.
517 "message": _truncate_for_database(
518 _exception_context(record) + record["message"]
519 ),
520 "module": record["name"],
521 "function": record["function"],
522 "line_no": int(record["line"]),
523 "level": record["level"].name,
524 "research_id": research_id,
525 "username": username,
526 }
528 # Queue unless we are on a thread where a synchronous SQLCipher write
529 # is actually safe — i.e. the main thread with NO event loop running
530 # on it (startup, CLI). Everything else goes to the background writer.
531 #
532 # The event-loop check is load-bearing and its absence was a migration
533 # regression. Flask's version of this guard read
534 # if not has_app_context() or current_thread().name != "MainThread"
535 # and the port dropped the first clause (correctly — there is no Flask)
536 # while keeping a thread-name test whose MEANING INVERTED. Under
537 # Werkzeug, request and Socket.IO handlers ran on "Thread-N", so the
538 # synchronous branch was effectively startup-only. Under uvicorn the
539 # event loop IS "MainThread" (web/app.py calls uvicorn.run() from
540 # main()), so every log line emitted on the loop took the synchronous
541 # branch: _write_log_to_database -> get_user_db_session -> potentially
542 # open_user_database, i.e. SQLCipher PBKDF2 key derivation, engine
543 # build and a commit, all on the event loop. With workers=1 that
544 # stalls the entire process.
545 #
546 # Thread name is the wrong question; "is a loop running here" is the
547 # right one, and it stays correct however the server is launched.
548 # Note no test could observe this: TestClient drives the app from an
549 # "asyncio-portal-*" thread, never "MainThread".
550 try:
551 asyncio.get_running_loop()
552 on_event_loop = True
553 except RuntimeError:
554 on_event_loop = False
556 if on_event_loop or threading.current_thread().name != "MainThread":
557 try:
558 _log_queue.put_nowait(log_entry)
559 except queue.Full:
560 pass # Drop log if queue is full to avoid blocking
561 else:
562 _write_log_to_database(log_entry)
563 finally:
564 _sink_state.in_sink = False
567def _truncate_for_database(message: str) -> str:
568 """Bound the persisted size of a log message.
570 ``DATABASE_MESSAGE_MAX_LENGTH`` is the preserved-prefix length;
571 the stored string is at most that plus a ~100-char suffix that
572 reports the original length so debug context is not lost. Full
573 messages remain available in container-log/stderr/file sinks.
574 """
575 if len(message) <= DATABASE_MESSAGE_MAX_LENGTH:
576 return message
577 suffix = (
578 f"… (truncated; full message in server logs; "
579 f"original length: {len(message)} chars)"
580 )
581 return message[:DATABASE_MESSAGE_MAX_LENGTH] + suffix
584def _exception_context(record) -> str:
585 """Render a short exception prefix from the log record, if any.
587 Loguru stores the active exception in ``record["exception"]`` as a
588 ``RecordException`` namedtuple (or None) populated by ``.exception()``
589 or by ``logger.opt(exception=...)``. Without this, the database sink
590 persists messages like ``"LangGraph agent error"`` with the actual
591 exception type and message only on stderr — investigators querying
592 ``app_logs`` see an empty ERROR row. The exception is always
593 separately available in container-log/stderr/file sinks; this just
594 gives the DB row enough context to triage. The full traceback is
595 deliberately NOT included because ``diagnose=False`` on the database
596 sink (see ``config_logger``) and the password-redaction policy
597 (#4182) forbid it for encrypted-DB persistence.
598 """
599 exc = record.get("exception") if record else None
600 if not exc:
601 return ""
602 # RecordException is a namedtuple of (type, value, traceback).
603 # ``value`` is the exception instance (or None if un-picklable).
604 exc_type = getattr(exc, "type", None)
605 exc_value = getattr(exc, "value", None)
606 if exc_type is None and isinstance(exc, tuple) and len(exc) >= 1:
607 exc_type = exc[0]
608 if exc_value is None and len(exc) >= 2: 608 ↛ 610line 608 didn't jump to line 610 because the condition on line 608 was always true
609 exc_value = exc[1]
610 if exc_type is None: 610 ↛ 611line 610 didn't jump to line 611 because the condition on line 610 was never true
611 return ""
612 type_name = getattr(exc_type, "__name__", str(exc_type))
613 if exc_value is None: 613 ↛ 614line 613 didn't jump to line 614 because the condition on line 613 was never true
614 return f"[{type_name}] "
615 try:
616 value_text = str(exc_value)
617 if len(value_text) > 4096:
618 value_text = value_text[:4096]
619 # Deferred import to avoid potential circular imports during module initialization
620 from ..security.log_sanitizer import sanitize_error_message
622 value_text = sanitize_error_message(value_text)
623 except Exception:
624 value_text = "<unprintable exception>"
625 if not value_text: 625 ↛ 626line 625 didn't jump to line 626 because the condition on line 625 was never true
626 return f"[{type_name}] "
627 if len(value_text) > 240:
628 value_text = value_text[:237] + "…"
629 return f"[{type_name}: {value_text}] "
632def _truncate_for_frontend(message: str) -> str:
633 """Bound the wire size of an outbound log message.
635 ``FRONTEND_MESSAGE_MAX_LENGTH`` caps the *preserved prefix* of the
636 original message. When truncation kicks in, a short indicator is
637 appended that names the original length and points the user at the
638 server-side logs for the full text, so the returned string is
639 ``FRONTEND_MESSAGE_MAX_LENGTH`` plus the fixed indicator overhead
640 (~100 bytes). Verbose diagnostic logs (e.g. ``[FETCH] page_text``
641 which inlines the full extracted page body) are useless in the UI
642 when displayed in full and inflate socket payloads + client-side
643 memory; container-log/stderr, file, and DB sinks remain unchanged.
644 """
645 if len(message) <= FRONTEND_MESSAGE_MAX_LENGTH: 645 ↛ 647line 645 didn't jump to line 647 because the condition on line 645 was always true
646 return message
647 suffix = (
648 f"… (truncated; full message in server logs; "
649 f"original length: {len(message)} chars)"
650 )
651 return message[:FRONTEND_MESSAGE_MAX_LENGTH] + suffix
654def frontend_progress_sink(message: loguru.Message) -> None:
655 """
656 Sink that sends messages to the frontend.
658 Args:
659 message: The log message to send.
661 """
662 record = message.record
663 research_id = _get_research_id(record)
664 if research_id is None:
665 # If we don't have a research ID, don't send anything.
666 # Can't use logger here as it causes deadlock
667 return
669 # Defence in depth (R4-09): never forward policy-audit log lines
670 # to WebSocket subscribers. They carry engine names + reason codes
671 # which could leak the active scope to a cross-origin observer under
672 # CORS=*. Today policy_audit logs don't bind research_id so the
673 # research_id guard above already skips them; this filter is the
674 # explicit guarantee in case a future call site binds both.
675 if record.get("extra", {}).get("policy_audit"):
676 return
678 # Whose research this is. Subscriptions are keyed by (owner, research_id)
679 # because a benchmark id is only unique within one user's database, so an
680 # emit without the owner reaches nobody. Resolved the same way
681 # ``database_sink`` resolves it: the bound ``username`` extra first, then
682 # the per-thread research context set when the research thread started.
683 username = record.get("extra", {}).get("username")
684 if not username:
685 ctx = _get_research_context_fallback()
686 if ctx: 686 ↛ 687line 686 didn't jump to line 687 because the condition on line 686 was never true
687 username = ctx.get("username")
688 if not username:
689 # Same third source as database_sink; see the note there.
690 username = _get_request_username()
691 if not username:
692 # Nothing to scope the emit to. Dropping is the fail-closed choice:
693 # the alternative (emit to every subscriber of this id) is exactly
694 # the cross-user leak the keying exists to prevent.
695 return
697 frontend_log = {
698 "log_entry": {
699 "message": _truncate_for_frontend(record["message"]),
700 "type": record["level"].name, # Keep original case
701 "time": record["time"].isoformat(),
702 },
703 }
704 emit_to_subscribers(
705 "research_progress",
706 research_id,
707 frontend_log,
708 owner=username,
709 enable_logging=False,
710 )
713def flush_log_queue():
714 """Drain all pending logs from the queue to the database.
716 Called from the FastAPI lifespan shutdown (see ``fastapi_app.py``)
717 to flush whatever the background daemon hadn't written before the
718 DB was closed. The request path doesn't call this — the
719 ``start_log_queue_processor`` daemon handles steady-state drainage
720 so requests never block on DB writes.
721 """
722 flushed = 0
723 while not _log_queue.empty():
724 try:
725 log_entry = _log_queue.get_nowait()
726 _write_log_to_database(log_entry)
727 flushed += 1
728 except queue.Empty:
729 break
730 except Exception:
731 pass # noqa: silent-exception — DB errors during log flush must not propagate
733 if flushed > 0:
734 logger.debug(f"Flushed {flushed} queued log entries to database")
737def start_log_queue_processor(app=None) -> threading.Thread:
738 """Start the background daemon that drains the log queue into the DB.
740 Idempotent: calling twice is a no-op (returns the existing thread).
741 Under FastAPI there's no Flask app context to push, so the daemon
742 just runs ``_process_log_queue`` directly. The ``app`` parameter is
743 accepted for backward compat with any legacy Flask call sites that
744 still pass it; it is otherwise ignored.
746 Returns:
747 The daemon thread (running).
748 """
749 global _queue_processor_thread
750 with _queue_processor_lock:
751 if (
752 _queue_processor_thread is not None
753 and _queue_processor_thread.is_alive()
754 ):
755 return _queue_processor_thread
757 _stop_queue.clear()
758 _queue_processor_thread = threading.Thread(
759 target=_process_log_queue,
760 name="log-queue-processor",
761 daemon=True,
762 )
763 _queue_processor_thread.start()
764 thread = _queue_processor_thread
765 logger.info("Log queue processor daemon started")
766 return thread
769def stop_log_queue_processor(timeout: float = 2.0) -> None:
770 """Signal the log queue processor to stop and wait briefly for it."""
771 global _queue_processor_thread
772 _stop_queue.set()
773 with _queue_processor_lock:
774 thread = _queue_processor_thread
775 if thread is not None:
776 thread.join(timeout=timeout)
777 # Only clear the reference if the thread actually exited. If join
778 # timed out the daemon is still running, and clearing the ref would
779 # let a subsequent start_log_queue_processor() spawn a second
780 # daemon that drains the same queue concurrently. Re-check identity
781 # under the lock so we don't accidentally null out a fresh thread
782 # that another start spawned in the meantime.
783 if not thread.is_alive():
784 with _queue_processor_lock:
785 if _queue_processor_thread is thread: 785 ↛ exitline 785 didn't jump to the function exit
786 _queue_processor_thread = None
789def config_logger(name: str, debug: bool = False) -> None:
790 """
791 Configures the default logger.
793 Args:
794 name: The name to use for the log file.
795 debug: Whether to enable unsafe debug logging.
797 """
798 from ..security.log_sanitizer import sanitize_log_record
800 logger.configure(patcher=sanitize_log_record)
802 logger.enable("local_deep_research")
803 logger.remove()
805 # Log to console (stderr), database, and frontend (Socket.IO).
806 # All three sinks track the stderr level — sending DEBUG logs to the
807 # DB and to every connected Socket.IO client in production would flood
808 # storage and the network with high-frequency internal events.
809 # `diagnose=False` on the DB/frontend sinks: loguru's `diagnose=True`
810 # dumps local variables on exceptions, which can contain request
811 # bodies, session data, or DB rows — we don't want those leaving the
812 # stderr sink.
813 sink_level = "DEBUG" if debug else "INFO"
815 # loguru's diagnose=True renders repr() of every local variable in every
816 # traceback frame on exceptions. Under LDR_APP_DEBUG that would dump
817 # credentials living in frame locals (api_key, SQLCipher password,
818 # Authorization headers) into the stderr sink. Gate diagnose behind a
819 # separate explicit opt-in so enabling LDR_APP_DEBUG for general debug
820 # output does not also enable localvar dumps. Default OFF even when debug
821 # is on, and never enabled on the DB/frontend sinks.
822 #
823 # security/secure_logging.py gates provider/engine exception tracebacks
824 # on the same two flags (its is_diagnose_mode()). Deliberate divergence:
825 # the wrapper reads LDR_APP_DEBUG from the environment only, while the
826 # ``debug`` argument here may also come from the app.debug DB setting /
827 # legacy server_config.json — so DB-enabled debug affects this sink's
828 # diagnose but never the wrapper's traceback gate. Kept inline (not
829 # env_truthy) because log_utils must not import security at module
830 # level: security/__init__.py runs install_audit_hook() on import.
831 diagnose = debug and os.environ.get(
832 "LDR_LOGURU_DIAGNOSE", ""
833 ).strip().lower() in ("1", "true", "yes")
835 # ``diagnose`` renders the repr() of every frame-local — which can
836 # include the SQLCipher master password and other credentials — into
837 # the rendered exception block. Allow it ONLY on the ephemeral, local
838 # stderr sink the operator explicitly opted into. The database sink
839 # PERSISTS into the user's own encrypted DB and the frontend sink
840 # SHIPS to the browser, so they must NEVER render frame locals, even
841 # under LDR_LOGURU_DIAGNOSE. This single chokepoint protects every
842 # credential-bearing exception handler app-wide against the
843 # frame-locals leak, independent of per-site logging discipline
844 # (#4182).
845 #
846 # enqueue=True on stderr: loguru hands the formatted record to a
847 # multiprocessing.SimpleQueue and a dedicated background thread does the
848 # actual stderr write, decoupling a log call from stderr I/O for up to
849 # one queue pipe buffer (~64KB on Linux) rather than never blocking.
850 # Without it, when stderr back-pressures (e.g. a slow/full `docker logs`
851 # pipe in CI) the lock-holder blocks mid-write and ALL logging threads
852 # pile up behind the lock, freezing the request pipeline for ~60s
853 # (#4431). Captured forensically: 3/5 server threads parked in loguru's
854 # _protected_lock under load. This is a bounded grace window, not
855 # immunity: SimpleQueue.put() still runs inside Handler.emit's
856 # _protected_lock, and its OS-pipe send_bytes() blocks once that buffer
857 # fills — e.g. if the background writer thread itself stalls on a
858 # wedged stderr long enough to drain nothing, reproducing the same
859 # pile-up (loguru/_handler.py Handler.emit / _queued_writer,
860 # multiprocessing/queues.py SimpleQueue.put). The database/progress
861 # sinks keep their own emitting-thread context capture and are left
862 # synchronous.
863 logger.add(sys.stderr, level=sink_level, diagnose=diagnose, enqueue=True)
864 logger.add(database_sink, level=sink_level, diagnose=False)
865 logger.add(frontend_progress_sink, level=sink_level, diagnose=False)
867 if debug:
868 logger.warning(
869 "DEBUG logging is enabled (LDR_APP_DEBUG=true). "
870 "Logs may contain sensitive data (queries, answers, API responses). "
871 "Do NOT use in production."
872 )
874 if diagnose:
875 logger.warning(
876 "LDR_LOGURU_DIAGNOSE is enabled: exception tracebacks will include "
877 "local variable values, which may contain credentials (API keys, "
878 "passwords, tokens). Do NOT use in production."
879 )
881 # Optionally log to file if enabled (disabled by default for security)
882 # Check environment variable first, then database setting
883 enable_file_logging = (
884 os.environ.get("LDR_ENABLE_FILE_LOGGING", "").lower() == "true"
885 )
887 # File logging is controlled only by environment variable for simplicity
888 # Database settings are not available at logger initialization time
890 if enable_file_logging:
891 log_file = _LOG_DIR / f"{name}.log"
892 logger.add(
893 log_file,
894 level="DEBUG",
895 rotation="10 MB",
896 retention="7 days",
897 compression="zip",
898 # diagnose=False: the file sink is persistent and unencrypted,
899 # so — like the DB and frontend sinks — it must never render
900 # frame-local credentials, even under LDR_LOGURU_DIAGNOSE.
901 # Frame-local dumps go only to the ephemeral stderr sink, which
902 # is what the policy comment above describes (#4182).
903 diagnose=False,
904 )
905 logger.warning(
906 f"File logging enabled - logs will be written to {log_file}. "
907 "WARNING: Log files are unencrypted and may contain sensitive data!"
908 )
910 # Add a special log level for milestones. Also registered at import
911 # (see _register_milestone_level) so processes that never call this
912 # function still have it; kept here for the re-entrant case.
913 _register_milestone_level()