Coverage for src/local_deep_research/utilities/log_utils.py: 87%
301 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +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 / agent log lines can carry large payloads
87# (model outputs, long tracebacks, extracted text) — many KB per call — which
88# is useless 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 do not apply this cap, so longer text stays available for grep/DB
91# queries -- up to the 32 KiB per-string bound the redaction patcher
92# (``_sanitize_record``) applies to every sink. The cap bounds the
93# *prefix* preserved from the original message; the wire payload is the
94# prefix plus a short truncation indicator (~100 bytes), so it can exceed
95# this value by that fixed overhead.
96FRONTEND_MESSAGE_MAX_LENGTH = 2000
98# Cap the size of messages persisted to ResearchLog. The DB column is
99# unbounded TEXT, so a long langgraph run can accumulate thousands of
100# 10 KB rows — paginated reads (PR #4037) hide the symptom but don't
101# stop the storage growth. Same prefix-plus-indicator semantics as
102# FRONTEND_MESSAGE_MAX_LENGTH; container-log/stderr/file sinks do not
103# apply this cap, so text up to the patcher's 32 KiB per-string bound is
104# still available out-of-band.
105DATABASE_MESSAGE_MAX_LENGTH = 5000
108class InterceptHandler(logging.Handler):
109 """
110 Intercepts logging messages and forwards them to Loguru's logger.
111 """
113 def emit(self, record: logging.LogRecord) -> None:
114 # Get corresponding Loguru level if it exists.
115 try:
116 level: str | int = logger.level(record.levelname).name
117 except ValueError:
118 level = record.levelno
120 # Find caller from where originated the logged message.
121 frame, depth = inspect.currentframe(), 0
122 while frame: 122 ↛ 131line 122 didn't jump to line 131 because the condition on line 122 was always true
123 filename = frame.f_code.co_filename
124 is_logging = filename == logging.__file__
125 is_frozen = "importlib" in filename and "_bootstrap" in filename
126 if depth > 0 and not (is_logging or is_frozen):
127 break
128 frame = frame.f_back
129 depth += 1
131 logger.opt(depth=depth, exception=record.exc_info).log(
132 level, record.getMessage()
133 )
136def log_for_research(
137 to_wrap: Callable[[str, ...], Any],
138) -> Callable[[str, ...], Any]:
139 """
140 Decorator for a function that's part of the research process. It expects the function to
141 take the research ID (UUID) as the first parameter, and configures all log
142 messages made during this request to include the research ID.
144 Args:
145 to_wrap: The function to wrap. Should take the research ID as the first parameter.
147 Returns:
148 The wrapped function.
150 """
152 @wraps(to_wrap)
153 def wrapped(research_id: str, *args: Any, **kwargs: Any) -> Any:
154 cv_handle = _research_id_var.set(research_id)
155 try:
156 return to_wrap(research_id, *args, **kwargs)
157 finally:
158 _research_id_var.reset(cv_handle)
160 return wrapped
163def _get_research_context_fallback() -> dict | None:
164 """Read the per-thread research context, if any.
166 Used as a fallback when individual log calls don't bind research_id/
167 username via ``logger.bind``. The research thread sets this once at
168 startup via ``set_search_context``, so every subsequent log call from
169 the same thread picks up research_id, username, and user_password
170 automatically — without requiring every call site to remember to bind.
171 """
172 try:
173 from .thread_context import get_search_context
175 return get_search_context()
176 except Exception:
177 return None
180def _get_request_username() -> str | None:
181 """Read the request-scoped username, if any.
183 The FastAPI successor to main's ``flask.session.get("username")``
184 fallback in the log sinks. Reads the contextvar ``DatabaseMiddleware``
185 populates per request (and that background workers push explicitly via
186 ``request_user``), so a log call that binds ``research_id`` without a
187 ``username`` -- as report_assembly_service does from a real request
188 thread -- still resolves an owner instead of being dropped.
190 Imported lazily and never allowed to raise: this runs inside a loguru
191 sink, where an exception would break logging itself.
192 """
193 try:
194 from .request_context import get_current_username
196 return get_current_username()
197 except Exception:
198 return None
201def _get_research_id(record=None) -> str | None:
202 """
203 Gets the current research ID (UUID), if present.
205 Args:
206 record: Optional loguru record that might contain bound research_id
208 Returns:
209 The current research ID (UUID), or None if it does not exist.
210 """
211 # First check if research_id is bound to the log record
212 if record and "extra" in record and "research_id" in record["extra"]:
213 return record["extra"]["research_id"]
214 # Then check the contextvar (set by @log_for_research on the running
215 # task/thread)
216 rid = _research_id_var.get()
217 if rid:
218 return rid
219 # Fall back to per-thread research context — research-thread logger
220 # calls without an explicit bind still get attributed correctly.
221 ctx = _get_research_context_fallback()
222 if ctx:
223 return ctx.get("research_id")
224 return None
227# Counters for swallowed exceptions in the logging path. Bare except: pass
228# is required here (logging errors must not propagate or recurse), but we
229# write a stderr line on each new occurrence so silent failures aren't
230# invisible — only active when LDR_APP_DEBUG=true so production stderr
231# stays clean.
232_silent_exc_counts: dict[str, int] = {}
235def _report_silent_exception(
236 where: str,
237 exc_type_name: str,
238 username: str | None = None,
239 research_id: str | None = None,
240 level: str | None = None,
241) -> None:
242 """Surface a swallowed logging-path exception to stderr.
244 Bypasses ``logger`` to avoid recursing back through ``database_sink``.
245 Rate-limited to first occurrence + every 100th repeat for the same
246 ``where`` key, so a persistent failure mode doesn't flood the console.
248 Note: takes the exception's TYPE NAME as a plain string (not the
249 exception object). The caller does ``type(exc).__name__`` and passes
250 the result. This is deliberate — CodeQL's taint analyzer treats any
251 function frame holding a ``BaseException`` captured from a password-
252 bearing call site as tainted, and flags every stderr write inside
253 that frame. Receiving only a type-name string severs the flow at
254 the boundary.
255 """
256 if os.environ.get("LDR_APP_DEBUG", "").lower() not in ("1", "true", "yes"): 256 ↛ 258line 256 didn't jump to line 258 because the condition on line 256 was always true
257 return
258 n = _silent_exc_counts.get(where, 0) + 1
259 _silent_exc_counts[where] = n
260 if n != 1 and n % 100 != 0:
261 return
262 parts = []
263 if username is not None:
264 parts.append(f"username={username!r}")
265 if research_id is not None:
266 parts.append(f"research_id={research_id!r}")
267 if level is not None:
268 parts.append(f"level={level!r}")
269 ctx = " ".join(parts)
270 # CodeQL's py/clear-text-logging-sensitive-data may flag this stderr
271 # write because the function frame is reachable from the logging path.
272 # (_write_log_to_database no longer holds any credential at all since
273 # #5538 -- it binds to the already-open engine.) The data
274 # actually written is only plain strings — `where` (literal),
275 # `exc_type_name` (`type(exc).__name__`), and `username/research_id/level`
276 # repr'd from the queue entry. No password value ever reaches the
277 # formatter; the helper signature deliberately accepts only typed
278 # primitives. If CodeQL flags this line, dismiss the alert as a
279 # false positive in the Security tab with that justification.
280 sys.stderr.write(
281 f"[log-utils] {where} swallowed (count={n}): "
282 f"{exc_type_name}{(' ' + ctx) if ctx else ''}\n"
283 )
284 sys.stderr.flush()
287def _clear_daemon_thread_credentials() -> None:
288 """Drop any DB session / cached credential held by the current thread.
290 Invoked at the end of every log-queue drain iteration. The queue
291 daemon is a long-lived ``daemon=True`` thread whose credentials the
292 dead-thread sweeper (``cleanup_dead_threads``) can never reclaim,
293 because the thread stays alive for the whole process. Clearing here
294 guarantees the ``log-queue-processor`` thread never retains a
295 plaintext credential in ``thread_local_session``'s
296 ``_thread_credentials``.
298 Best-effort: must never raise, or a single cleanup hiccup would crash
299 the daemon and silently stop all subsequent log persistence.
300 """
301 try:
302 from ..database.thread_local_session import cleanup_current_thread
304 cleanup_current_thread()
305 except Exception:
306 pass # noqa: silent-exception — cleanup must not kill the log daemon
309def _process_log_queue():
310 """Process logs from the queue in a dedicated background thread.
312 Under FastAPI there's no Flask app context to gate on — we just
313 drain the queue and write entries directly. Main gated this on
314 ``has_app_context()`` and re-queued otherwise; with no such context
315 here the write is unconditional, which is safe because
316 ``_write_log_to_database`` now refuses to open a closed database.
318 Safe to run off the main thread: the write binds a session to the
319 user's already-open engine, and the underlying SQLite engines are
320 opened with ``check_same_thread=False``. It never reopens a closed
321 database, so a post-logout backlog can't resurrect a user's
322 connection from this thread (#5538). Errors are swallowed rather
323 than crashing the processor.
324 """
325 while not _stop_queue.is_set():
326 try:
327 log_entry = _log_queue.get(timeout=0.1)
328 if log_entry is None:
329 continue
330 try:
331 _write_log_to_database(log_entry)
332 finally:
333 # Belt-and-braces credential hygiene (#5538). This daemon is
334 # a long-lived daemon=True thread, so the dead-thread
335 # credential sweeper (cleanup_dead_threads) never reclaims
336 # it. Clear any per-thread DB session / cached credential at
337 # the end of every drain iteration so the
338 # log-queue-processor thread can never accumulate a
339 # plaintext credential in _thread_credentials.
340 _clear_daemon_thread_credentials()
341 except queue.Empty:
342 continue
343 except Exception as exc:
344 # noqa: silent-exception — must not let logging errors crash the log processor thread.
345 # Wrap the report itself: if stderr is closed (broken pipe etc.)
346 # an IOError from inside an except handler propagates and would
347 # kill the daemon thread, silently breaking all subsequent log
348 # persistence for the rest of the process lifetime.
349 try:
350 _report_silent_exception(
351 "process_log_queue", type(exc).__name__
352 )
353 except Exception:
354 pass # noqa: silent-exception — broken stderr must not kill the daemon
357def _write_log_to_database(log_entry: dict) -> None:
358 """Persist one queued log entry into the user's already-open database.
360 The queue drain must NEVER reopen a closed database. A log entry that
361 was queued while a user was active can be drained by the background
362 daemon *after* that user has logged out; reopening their encrypted DB
363 here would silently re-add the connection (``is_user_connected`` would
364 flip back to True) and re-cache their credentials — resurrecting
365 session state with no user action and no attacker involved.
367 So we gate strictly on an already-open connection
368 (``db_manager.is_user_connected``) and bind to the EXISTING engine via
369 ``db_manager.get_session`` — which never opens a new database and needs
370 no password. Entries for a user with no live connection (e.g. a
371 post-logout backlog) are dropped; the full message is still available
372 in the stderr/file sinks.
374 In-flight logging is preserved: while a user is active — including a
375 running research job, which keeps their DB open — the connection stays
376 live, so their logs continue to be written normally.
377 """
378 from ..database.encrypted_db import db_manager
380 username = log_entry.get("username")
381 if not username:
382 return
384 # Gate on an already-open connection. Do NOT lazily reopen the DB:
385 # after logout the connection is gone, and reopening it here would
386 # resurrect the user's connected state from a stale queued backlog.
387 if not db_manager.is_user_connected(username):
388 return
390 db_session = None
391 try:
392 # Binds a session to the ALREADY-open engine. Returns None (never
393 # reopens) if the connection was closed in the small race between
394 # the is_user_connected check above and this call; a rekey sidecar
395 # fault in that window raises instead and is reported below like
396 # any other failed write (the entry is dropped). We can't use
397 # get_user_db_session here: it would resolve+cache a password and
398 # lazily reopen the DB, which is exactly the post-logout
399 # resurrection this fix removes. The session is explicitly closed
400 # in the finally block below, so the QueuePool connection is
401 # returned and no FD leaks.
402 db_session = db_manager.get_session(username) # noqa: raw-session
403 if db_session is not None: 403 ↛ 434line 403 didn't jump to line 434 because the condition on line 403 was always true
404 db_log = ResearchLog(
405 timestamp=log_entry["timestamp"],
406 message=log_entry["message"],
407 module=log_entry["module"],
408 function=log_entry["function"],
409 line_no=log_entry["line_no"],
410 level=log_entry["level"],
411 research_id=log_entry["research_id"],
412 )
413 db_session.add(db_log)
414 db_session.commit()
415 except Exception as exc:
416 # noqa: silent-exception — DB errors in the logging path must not propagate or recurse.
417 # Wrap the report itself so a broken-stderr IOError can't escape and
418 # be re-caught by an outer logging-aware handler somewhere upstream.
419 try:
420 _report_silent_exception(
421 "write_log_to_database",
422 type(exc).__name__,
423 username=log_entry.get("username"),
424 research_id=log_entry.get("research_id"),
425 level=log_entry.get("level"),
426 )
427 except Exception:
428 pass # noqa: silent-exception — broken stderr must not bubble out of logging path
429 finally:
430 # The session is freshly created per drain, bound to the shared
431 # per-user engine — close it so its pooled connection is returned
432 # (close() also rolls back any pending transaction from a failed
433 # commit). Never let a close error escape the logging path.
434 if db_session is not None:
435 try:
436 db_session.close()
437 except Exception:
438 pass # noqa: silent-exception — session close failure must not propagate
441def database_sink(message: loguru.Message) -> None:
442 """
443 Sink that saves messages to the database.
444 Queues logs from background threads for later processing.
446 Args:
447 message: The log message to save.
449 """
450 # Thread-local re-entry guard. Any code path inside this sink (or
451 # `_write_log_to_database`) that happens to call a logger — including
452 # loguru's own sink-error handler when `catch=True` is set — would
453 # re-invoke this function and recurse. The `try/except Exception: pass`
454 # in `_write_log_to_database` catches most cases, but a nested sink
455 # error handler can still land here. Fail closed for the nested call.
456 if getattr(_sink_state, "in_sink", False): 456 ↛ 457line 456 didn't jump to line 457 because the condition on line 456 was never true
457 return
458 _sink_state.in_sink = True
459 try:
460 record = message.record
461 research_id = _get_research_id(record)
463 # Resolve the username the log belongs to. The queue daemon thread
464 # can't read the research thread's ContextVar storage, so we capture
465 # it here in the emitting thread.
466 #
467 # We deliberately do NOT capture the user's password. The queue entry
468 # must never carry a plaintext credential: the background drain would
469 # otherwise use it to reopen the user's encrypted DB after logout,
470 # reinstating their connected state (#5538). The drain writes only
471 # when the DB is already open -- see _write_log_to_database.
472 #
473 # Source priority:
474 # 1. logger.bind(...) extras on the record itself
475 # 2. per-thread research context (set once when the research thread
476 # starts, so every subsequent log call inherits it without
477 # requiring an explicit bind)
478 username = record.get("extra", {}).get("username")
479 ctx = _get_research_context_fallback()
480 if ctx and not username: 480 ↛ 481line 480 didn't jump to line 481 because the condition on line 480 was never true
481 username = ctx.get("username")
482 if research_id is not None and not username:
483 # Third source, ported from main: it resolved the request user
484 # via `flask.session.get("username")` when a research_id was
485 # bound but no username was. That fallback is not dead code --
486 # report_assembly_service binds research_id alone from a real
487 # request thread, and without this the entry resolves to
488 # username=None and is silently dropped by the `if not username`
489 # guard downstream. get_current_username() is the direct
490 # successor (the contextvar DatabaseMiddleware populates).
491 #
492 # The `research_id is not None` condition is main's, and it is
493 # load-bearing for data retention, not just noise control.
494 # DatabaseMiddleware populates that contextvar for EVERY
495 # authenticated request, so without the gate any INFO record
496 # emitted during any request resolves an owner, clears the skip
497 # guard below, and is persisted to that user's app_logs with
498 # research_id = NULL. Those rows are unreachable by the only
499 # mechanism that ever removes app_logs -- the ON DELETE CASCADE
500 # from research_history -- so neither "delete this research" nor
501 # "clear all history" can touch them, and there is no retention
502 # job. They would accumulate permanently with no way to delete
503 # them. main's comment put it as: "ResearchLog is research-scoped
504 # by design -- auth and other system DEBUG logs (research_id=None)
505 # don't belong there."
506 username = _get_request_username()
508 # Skip persistence for system logs that have no research context.
509 # These can't be written to any per-user encrypted DB and would just
510 # churn through the queue + daemon for no useful end state.
511 if research_id is None and username is None:
512 return
514 # Create log entry dict
515 log_entry = {
516 "timestamp": record["time"],
517 # Prefix the message with exception type/value when the record
518 # carries one (logger.exception / logger.opt(exception=...)).
519 # Keeps the full traceback off the encrypted-DB row (diagnose=False
520 # on this sink; password-redaction policy #4182) while still
521 # leaving the database log useful for triage.
522 "message": _truncate_for_database(
523 _exception_context(record) + record["message"]
524 ),
525 "module": record["name"],
526 "function": record["function"],
527 "line_no": int(record["line"]),
528 "level": record["level"].name,
529 "research_id": research_id,
530 "username": username,
531 }
533 # Queue unless we are on a thread where a synchronous SQLCipher write
534 # is actually safe — i.e. the main thread with NO event loop running
535 # on it (startup, CLI). Everything else goes to the background writer.
536 #
537 # The event-loop check is load-bearing and its absence was a migration
538 # regression. Flask's version of this guard read
539 # if not has_app_context() or current_thread().name != "MainThread"
540 # and the port dropped the first clause (correctly — there is no Flask)
541 # while keeping a thread-name test whose MEANING INVERTED. Under
542 # Werkzeug, request and Socket.IO handlers ran on "Thread-N", so the
543 # synchronous branch was effectively startup-only. Under uvicorn the
544 # event loop IS "MainThread" (web/app.py calls uvicorn.run() from
545 # main()), so every log line emitted on the loop took the synchronous
546 # branch: _write_log_to_database -> get_user_db_session -> potentially
547 # open_user_database, i.e. SQLCipher PBKDF2 key derivation, engine
548 # build and a commit, all on the event loop. With workers=1 that
549 # stalls the entire process.
550 #
551 # Thread name is the wrong question; "is a loop running here" is the
552 # right one, and it stays correct however the server is launched.
553 # Note no test could observe this: TestClient drives the app from an
554 # "asyncio-portal-*" thread, never "MainThread".
555 try:
556 asyncio.get_running_loop()
557 on_event_loop = True
558 except RuntimeError:
559 on_event_loop = False
561 if on_event_loop or threading.current_thread().name != "MainThread":
562 try:
563 _log_queue.put_nowait(log_entry)
564 except queue.Full:
565 pass # Drop log if queue is full to avoid blocking
566 else:
567 _write_log_to_database(log_entry)
568 finally:
569 _sink_state.in_sink = False
572def _truncate_for_database(message: str) -> str:
573 """Bound the persisted size of a log message.
575 ``DATABASE_MESSAGE_MAX_LENGTH`` is the preserved-prefix length;
576 the stored string is at most that plus a ~100-char suffix that
577 reports the original length so debug context is not lost. The
578 container-log/stderr/file sinks keep up to the patcher's 32 KiB
579 per-string bound.
580 """
581 if len(message) <= DATABASE_MESSAGE_MAX_LENGTH:
582 return message
583 return message[:DATABASE_MESSAGE_MAX_LENGTH] + _truncation_suffix(message)
586def _truncation_suffix(message: str) -> str:
587 """Indicator appended by the DB and frontend caps.
589 The reported length counts what the patcher's 32 KiB bound already
590 dropped (its omission marker records the count), so it is the length
591 before any truncation; the server logs hold at most the first
592 ``_LOG_REDACTION_MAX_CHARS`` characters.
593 """
594 # Deferred import to avoid potential circular imports during module initialization
595 from ..security.log_sanitizer import (
596 _LOG_REDACTION_MAX_CHARS,
597 log_length_before_truncation,
598 )
600 return (
601 f"… (truncated; server logs keep up to {_LOG_REDACTION_MAX_CHARS} "
602 f"chars; original length: {log_length_before_truncation(message)} "
603 f"chars)"
604 )
607def _exception_context(record) -> str:
608 """Render a short exception prefix from the log record, if any.
610 Loguru stores the active exception in ``record["exception"]`` as a
611 ``RecordException`` namedtuple (or None) populated by ``.exception()``
612 or by ``logger.opt(exception=...)``. Without this, the database sink
613 persists messages like ``"LangGraph agent error"`` with the actual
614 exception type and message only on stderr — investigators querying
615 ``app_logs`` see an empty ERROR row. The exception is always
616 separately available in container-log/stderr/file sinks; this just
617 gives the DB row enough context to triage. The full traceback is
618 deliberately NOT included because ``diagnose=False`` on the database
619 sink (see ``config_logger``) and the password-redaction policy
620 (#4182) forbid it for encrypted-DB persistence.
621 """
622 exc = record.get("exception") if record else None
623 if not exc:
624 return ""
625 # RecordException is a namedtuple of (type, value, traceback).
626 # ``value`` is the exception instance (or None if un-picklable).
627 exc_type = getattr(exc, "type", None)
628 exc_value = getattr(exc, "value", None)
629 if exc_type is None and isinstance(exc, tuple) and len(exc) >= 1:
630 exc_type = exc[0]
631 if exc_value is None and len(exc) >= 2: 631 ↛ 633line 631 didn't jump to line 633 because the condition on line 631 was always true
632 exc_value = exc[1]
633 if exc_type is None: 633 ↛ 634line 633 didn't jump to line 634 because the condition on line 633 was never true
634 return ""
635 # Deferred import to avoid potential circular imports during module initialization
636 from ..security.log_sanitizer import (
637 _redact_additional_shapes,
638 sanitize_error_message,
639 type_name_for_log,
640 )
642 # ``type`` is the original class unless the patcher had to replace it
643 # (``security.log_sanitizer._record_type_is_safe``). Its name is read
644 # through ``type``'s own descriptor (never a metaclass's ``__name__``)
645 # and redacted, as in the patcher's rebuilt
646 # ``RuntimeError("<TypeName>: ...")`` text.
647 type_name = type_name_for_log(exc_type)
648 if exc_value is None: 648 ↛ 649line 648 didn't jump to line 649 because the condition on line 648 was never true
649 return f"[{type_name}] "
650 truncated = False
651 try:
652 value_text = str(exc_value)
653 # The patcher renders an application exception as a rebuilt
654 # ``RuntimeError("<TypeName>: ...")`` (``RuntimeError("<TypeName>")``
655 # when its text is empty) while ``type`` stays the original class;
656 # drop the repeated name so the row does not read
657 # ``[TypeName: TypeName: ...]`` or ``[TypeName: TypeName]``.
658 repeated = "{}: ".format(type_name)
659 if type(exc_value) is not exc_type:
660 if value_text.startswith(repeated):
661 value_text = value_text[len(repeated) :]
662 elif value_text == type_name: 662 ↛ 664line 662 didn't jump to line 664 because the condition on line 662 was always true
663 value_text = ""
664 if len(value_text) > 4096:
665 value_text = value_text[:4096]
666 # The credential shapes the log-record path adds on top of
667 # sanitize_error_message (``Api-Key:`` headers, slash-bearing DSN
668 # passwords, ``;``/``,`` credential parameters, quoted credential
669 # keys) run on the 4096-bounded text BEFORE the 240-character cap:
670 # a value the cap cuts below a pattern's minimum length (or a DSN
671 # cut before its '@') would no longer be recognised, and its head
672 # would print. This prefix goes into the DB row with no other
673 # redaction pass after it.
674 value_text = _redact_additional_shapes(
675 sanitize_error_message(value_text)
676 )
677 if len(value_text) > 240:
678 # And once more on the capped text (before the ellipsis, which
679 # an unterminated quoted value would otherwise take): these
680 # patterns only ever replace part of their input with a
681 # marker, so a second pass can hide more, never less.
682 value_text = _redact_additional_shapes(value_text[:237])
683 truncated = True
684 except Exception:
685 value_text = "<unprintable exception>"
686 if not value_text:
687 return f"[{type_name}] "
688 # The name and the text are each clean, but a credential pattern can
689 # span the two (a class named ``InvalidAuthorization`` and a bare token
690 # read as an ``Authorization:`` header): sanitize the joined line.
691 line = _redact_additional_shapes(
692 sanitize_error_message(f"{type_name}: {value_text}")
693 )
694 cap = len(type_name) + 2 + 240
695 if len(line) > cap - int(truncated): 695 ↛ 696line 695 didn't jump to line 696 because the condition on line 695 was never true
696 line = _redact_additional_shapes(line[: cap - 1])
697 truncated = True
698 if truncated:
699 line += "…"
700 return f"[{line}] "
703def _truncate_for_frontend(message: str) -> str:
704 """Bound the wire size of an outbound log message.
706 ``FRONTEND_MESSAGE_MAX_LENGTH`` caps the *preserved prefix* of the
707 original message. When truncation kicks in, a short indicator is
708 appended that names the original length and points the user at the
709 server-side logs, which keep up to 32 KiB of it, so the returned string is
710 ``FRONTEND_MESSAGE_MAX_LENGTH`` plus the fixed indicator overhead
711 (~100 bytes). Verbose diagnostic / agent logs (model outputs, long
712 tracebacks, extracted text) are useless in the UI when displayed in
713 full and inflate socket payloads + client-side memory;
714 container-log/stderr, file, and DB sinks do not apply this cap.
715 """
716 if len(message) <= FRONTEND_MESSAGE_MAX_LENGTH:
717 return message
718 return message[:FRONTEND_MESSAGE_MAX_LENGTH] + _truncation_suffix(message)
721def frontend_progress_sink(message: loguru.Message) -> None:
722 """
723 Sink that sends messages to the frontend.
725 Args:
726 message: The log message to send.
728 """
729 record = message.record
730 research_id = _get_research_id(record)
731 if research_id is None:
732 # If we don't have a research ID, don't send anything.
733 # Can't use logger here as it causes deadlock
734 return
736 # Defence in depth (R4-09): never forward policy-audit log lines
737 # to WebSocket subscribers. They carry engine names + reason codes
738 # which could leak the active scope to a cross-origin observer under
739 # CORS=*. Today policy_audit logs don't bind research_id so the
740 # research_id guard above already skips them; this filter is the
741 # explicit guarantee in case a future call site binds both.
742 if record.get("extra", {}).get("policy_audit"):
743 return
745 # Whose research this is. Subscriptions are keyed by (owner, research_id)
746 # because a benchmark id is only unique within one user's database, so an
747 # emit without the owner reaches nobody. Resolved the same way
748 # ``database_sink`` resolves it: the bound ``username`` extra first, then
749 # the per-thread research context set when the research thread started.
750 username = record.get("extra", {}).get("username")
751 if not username:
752 ctx = _get_research_context_fallback()
753 if ctx: 753 ↛ 754line 753 didn't jump to line 754 because the condition on line 753 was never true
754 username = ctx.get("username")
755 if not username:
756 # Same third source as database_sink; see the note there.
757 username = _get_request_username()
758 if not username:
759 # Nothing to scope the emit to. Dropping is the fail-closed choice:
760 # the alternative (emit to every subscriber of this id) is exactly
761 # the cross-user leak the keying exists to prevent.
762 return
764 frontend_log = {
765 "log_entry": {
766 "message": _truncate_for_frontend(record["message"]),
767 "type": record["level"].name, # Keep original case
768 "time": record["time"].isoformat(),
769 },
770 }
771 emit_to_subscribers(
772 "research_progress",
773 research_id,
774 frontend_log,
775 owner=username,
776 enable_logging=False,
777 )
780def flush_log_queue():
781 """Drain all pending logs from the queue to the database.
783 Called from the FastAPI lifespan shutdown (see ``fastapi_app.py``)
784 to flush whatever the background daemon hadn't written before the
785 DB was closed. The request path doesn't call this — the
786 ``start_log_queue_processor`` daemon handles steady-state drainage
787 so requests never block on DB writes.
788 """
789 flushed = 0
790 while not _log_queue.empty():
791 try:
792 log_entry = _log_queue.get_nowait()
793 _write_log_to_database(log_entry)
794 flushed += 1
795 except queue.Empty:
796 break
797 except Exception:
798 pass # noqa: silent-exception — DB errors during log flush must not propagate
800 if flushed > 0:
801 logger.debug(f"Flushed {flushed} queued log entries to database")
804def start_log_queue_processor(app=None) -> threading.Thread:
805 """Start the background daemon that drains the log queue into the DB.
807 Idempotent: calling twice is a no-op (returns the existing thread).
808 Under FastAPI there's no Flask app context to push, so the daemon
809 just runs ``_process_log_queue`` directly. The ``app`` parameter is
810 accepted for backward compat with any legacy Flask call sites that
811 still pass it; it is otherwise ignored.
813 Returns:
814 The daemon thread (running).
815 """
816 global _queue_processor_thread
817 with _queue_processor_lock:
818 if (
819 _queue_processor_thread is not None
820 and _queue_processor_thread.is_alive()
821 ):
822 return _queue_processor_thread
824 _stop_queue.clear()
825 _queue_processor_thread = threading.Thread(
826 target=_process_log_queue,
827 name="log-queue-processor",
828 daemon=True,
829 )
830 _queue_processor_thread.start()
831 thread = _queue_processor_thread
832 logger.info("Log queue processor daemon started")
833 return thread
836def stop_log_queue_processor(timeout: float = 2.0) -> None:
837 """Signal the log queue processor to stop and wait briefly for it."""
838 global _queue_processor_thread
839 _stop_queue.set()
840 with _queue_processor_lock:
841 thread = _queue_processor_thread
842 if thread is not None:
843 thread.join(timeout=timeout)
844 # Only clear the reference if the thread actually exited. If join
845 # timed out the daemon is still running, and clearing the ref would
846 # let a subsequent start_log_queue_processor() spawn a second
847 # daemon that drains the same queue concurrently. Re-check identity
848 # under the lock so we don't accidentally null out a fresh thread
849 # that another start spawned in the meantime.
850 if not thread.is_alive():
851 with _queue_processor_lock:
852 if _queue_processor_thread is thread: 852 ↛ exitline 852 didn't jump to the function exit
853 _queue_processor_thread = None
856def _sanitize_record(record) -> None:
857 """Global loguru patcher: scrub every record before it reaches a sink.
859 Runs once per record, ahead of ALL sinks (stderr, the encrypted per-user
860 DB, the browser progress stream, and the persistent, unencrypted
861 ``<LDR_DATA_DIR>/logs/*.log`` file), so it is the single chokepoint that
862 protects every sink regardless of per-call-site logging discipline:
864 * Control / format characters are stripped from the message (prevents log
865 forging via CR/LF, RLO, zero-width, etc.).
866 * Credential *shapes* and bare ``field=value`` secrets are redacted from
867 the message — the sink-level backstop for any call site that logged a
868 credential without going through ``scrub_error`` first.
869 * Structured ``extra`` / ``logger.bind(...)`` values are redacted by
870 sensitive key name AND by embedded credential shape, catching the
871 recurring family bug of a secret carried as a bound structured value
872 (a token/URL nested under a key the name-based checks miss).
873 * The *rendered* exception graph (value, notes, chained and grouped
874 exceptions) is scrubbed too, as a rebuilt copy, so a credential
875 embedded in an exception's string does not render verbatim to the
876 default-format stderr / file sinks — the message scrub alone never
877 reaches it.
879 All of this is ``security.log_sanitizer.redact_log_record``, which the
880 MCP subprocess installs as its own patcher too (``mcp/server.py``'s
881 ``configure_mcp_logging``); it fails closed per field. It is imported
882 lazily (never at this module's import time): importing ``..security``
883 runs ``install_audit_hook()``, and ``log_utils`` must stay importable
884 before the security layer is wired.
885 """
886 from ..security.log_sanitizer import redact_log_record
888 redact_log_record(record)
891def install_log_redaction() -> None:
892 """Install the shared credential-redaction patcher without adding sinks.
894 Entry points that configure their own sinks, such as the benchmark CLI,
895 use this before enabling their package logger. Importing this module
896 alone does not change the host process's Loguru configuration.
897 """
898 logger.configure(patcher=_sanitize_record)
901def config_logger(name: str, debug: bool = False) -> None:
902 """
903 Configures the default logger.
905 Args:
906 name: The name to use for the log file.
907 debug: Whether to enable unsafe debug logging.
909 """
910 install_log_redaction()
912 logger.enable("local_deep_research")
913 logger.remove()
915 # Log to console (stderr), database, and frontend (Socket.IO).
916 # All three sinks track the stderr level — sending DEBUG logs to the
917 # DB and to every connected Socket.IO client in production would flood
918 # storage and the network with high-frequency internal events.
919 # `diagnose=False` on the DB/frontend sinks: loguru's `diagnose=True`
920 # dumps local variables on exceptions, which can contain request
921 # bodies, session data, or DB rows — we don't want those leaving the
922 # stderr sink.
923 sink_level = "DEBUG" if debug else "INFO"
925 # loguru's diagnose=True renders repr() of every local variable in every
926 # traceback frame on exceptions. Under LDR_APP_DEBUG that would dump
927 # credentials living in frame locals (api_key, SQLCipher password,
928 # Authorization headers) into the stderr sink. Gate diagnose behind a
929 # separate explicit opt-in so enabling LDR_APP_DEBUG for general debug
930 # output does not also enable localvar dumps. Default OFF even when debug
931 # is on, and never enabled on the DB/frontend sinks.
932 #
933 # security/secure_logging.py gates provider/engine exception tracebacks
934 # on the same two flags (its is_diagnose_mode()). Deliberate divergence:
935 # the wrapper reads LDR_APP_DEBUG from the environment only, while the
936 # ``debug`` argument here may also come from the app.debug DB setting /
937 # legacy server_config.json — so DB-enabled debug affects this sink's
938 # diagnose but never the wrapper's traceback gate. Kept inline (not
939 # env_truthy) because log_utils must not import security at module
940 # level: security/__init__.py runs install_audit_hook() on import.
941 diagnose = debug and os.environ.get(
942 "LDR_LOGURU_DIAGNOSE", ""
943 ).strip().lower() in ("1", "true", "yes")
945 # ``diagnose`` renders the repr() of every frame-local — which can
946 # include the SQLCipher master password and other credentials — into
947 # the rendered exception block. Allow it ONLY on the ephemeral, local
948 # stderr sink the operator explicitly opted into. The database sink
949 # PERSISTS into the user's own encrypted DB and the frontend sink
950 # SHIPS to the browser, so they must NEVER render frame locals, even
951 # under LDR_LOGURU_DIAGNOSE. This single chokepoint protects every
952 # credential-bearing exception handler app-wide against the
953 # frame-locals leak, independent of per-site logging discipline
954 # (#4182).
955 #
956 # enqueue=True on stderr: loguru hands the formatted record to a
957 # multiprocessing.SimpleQueue and a dedicated background thread does the
958 # actual stderr write, decoupling a log call from stderr I/O for up to
959 # one queue pipe buffer (~64KB on Linux) rather than never blocking.
960 # Without it, when stderr back-pressures (e.g. a slow/full `docker logs`
961 # pipe in CI) the lock-holder blocks mid-write and ALL logging threads
962 # pile up behind the lock, freezing the request pipeline for ~60s
963 # (#4431). Captured forensically: 3/5 server threads parked in loguru's
964 # _protected_lock under load. This is a bounded grace window, not
965 # immunity: SimpleQueue.put() still runs inside Handler.emit's
966 # _protected_lock, and its OS-pipe send_bytes() blocks once that buffer
967 # fills — e.g. if the background writer thread itself stalls on a
968 # wedged stderr long enough to drain nothing, reproducing the same
969 # pile-up (loguru/_handler.py Handler.emit / _queued_writer,
970 # multiprocessing/queues.py SimpleQueue.put). The database/progress
971 # sinks keep their own emitting-thread context capture and are left
972 # synchronous.
973 logger.add(sys.stderr, level=sink_level, diagnose=diagnose, enqueue=True)
974 logger.add(database_sink, level=sink_level, diagnose=False)
975 logger.add(frontend_progress_sink, level=sink_level, diagnose=False)
977 if debug:
978 logger.warning(
979 "DEBUG logging is enabled (LDR_APP_DEBUG=true). "
980 "Logs may contain sensitive data (queries, answers, API responses). "
981 "Do NOT use in production."
982 )
984 if diagnose:
985 logger.warning(
986 "LDR_LOGURU_DIAGNOSE is enabled: exception tracebacks will include "
987 "local variable values, which may contain credentials (API keys, "
988 "passwords, tokens). Do NOT use in production."
989 )
991 # Optionally log to file if enabled (disabled by default for security)
992 # Check environment variable first, then database setting
993 enable_file_logging = (
994 os.environ.get("LDR_ENABLE_FILE_LOGGING", "").lower() == "true"
995 )
997 # File logging is controlled only by environment variable for simplicity
998 # Database settings are not available at logger initialization time
1000 if enable_file_logging:
1001 log_file = _LOG_DIR / f"{name}.log"
1002 logger.add(
1003 log_file,
1004 level="DEBUG",
1005 rotation="10 MB",
1006 retention="7 days",
1007 compression="zip",
1008 # diagnose=False: the file sink is persistent and unencrypted,
1009 # so — like the DB and frontend sinks — it must never render
1010 # frame-local credentials, even under LDR_LOGURU_DIAGNOSE.
1011 # Frame-local dumps go only to the ephemeral stderr sink, which
1012 # is what the policy comment above describes (#4182).
1013 diagnose=False,
1014 )
1015 logger.warning(
1016 f"File logging enabled - logs will be written to {log_file}. "
1017 "WARNING: Log files are unencrypted and may contain sensitive data!"
1018 )
1020 # Add a special log level for milestones. Also registered at import
1021 # (see _register_milestone_level) so processes that never call this
1022 # function still have it; kept here for the re-entrant case.
1023 _register_milestone_level()