learningcircuit--local-deep-research/tests/security/test_password_leakage.py

"""Tests that ``logger.exception`` sites in ``database/encrypted_db.py``,
``web/queue/processor_v2.py``, and ``scheduler/background.py`` never leak
the SQLCipher master password into log output.

Companion fix: issue #4182 (follow-up to #4131). API keys are rotatable,
but SQLCipher master passwords are not (TRUST.md §5) — a leaked password
in a CI log or shared error report forces the user to abandon their
encrypted DB.

The fix pattern, established in PRs #4168/#4175/#4181, is:

    except Exception as e:
        safe_msg = redact_secrets(str(e), password)
        logger.warning(f"... {safe_msg}")

These tests use ``loguru_caplog_full`` to capture the rendered exception
block (the existing ``loguru_caplog`` would false-pass on a leak that
lives only in the traceback frames). Each test is mutation-verified —
revert the production wrap and the test fails with the sentinel visible.
"""

import base64
from unittest.mock import MagicMock, patch
from urllib.parse import quote, quote_plus

import pytest


_LEAKED_PASSWORD = "ldr-master-pw-SHOULD-NEVER-LEAK-987654321"
_LEAKED_OLD_PASSWORD = "ldr-old-master-pw-ALSO-NEVER-LEAK-123"


def _all_encodings_of(secret: str) -> list:
    """Return every encoding the sentinel might appear under in a log.

    Mirrors the helper in ``test_api_key_leakage.py``. Extend when a
    fixed call site introduces a new transformation (e.g., bcrypt hash,
    JWT payload).
    """
    return [
        secret,
        quote(secret, safe=""),
        quote_plus(secret),
        repr(secret)[1:-1],
        base64.b64encode(secret.encode()).decode(),
        secret[:8],
    ]


def _assert_no_leak(text: str, secret: str, where: str) -> None:
    """Helper: assert no encoding of *secret* appears in *text*."""
    for encoding in _all_encodings_of(secret):
        assert encoding not in text, (
            f"password leaked at {where} as encoding {encoding!r}. The "
            f"except handler must call redact_secrets(str(e), password) "
            f"and use logger.warning (not logger.exception)."
        )


# ---------------------------------------------------------------------------
# database/encrypted_db.py
# ---------------------------------------------------------------------------


class TestEncryptedDBPasswordLeakage:
    """``DatabaseManager`` lifecycle methods receive the user's SQLCipher
    master password and pass it through to ``set_sqlcipher_key`` /
    ``create_sqlcipher_connection`` / ``BackupService``. An exception in
    any of those paths can carry frame locals containing the plaintext
    password under loguru ``diagnose=True``. The except handlers must
    drop the traceback chain and redact str(e).
    """

    def test_change_password_does_not_leak_either_password(
        self, loguru_caplog_full, tmp_path
    ):
        """``change_password`` has BOTH ``old_password`` and
        ``new_password`` in lexical scope. A failure in
        ``open_user_database`` or ``set_sqlcipher_rekey`` whose exception
        message embeds either password must not surface to logs.
        """
        from local_deep_research.database.encrypted_db import DatabaseManager

        manager = DatabaseManager.__new__(DatabaseManager)
        manager.has_encryption = True
        manager.connections = {}
        manager._connections_lock = MagicMock()
        manager._connections_lock.__enter__ = lambda *a: None
        manager._connections_lock.__exit__ = lambda *a: None

        # Drive _get_user_db_path -> a real, existing path so the
        # ``if not db_path.exists()`` short-circuit doesn't fire.
        fake_db = tmp_path / "fake.db"
        fake_db.touch()
        manager._get_user_db_path = lambda u: fake_db
        manager.close_user_database = lambda u: None

        # Construct an exception whose str() embeds BOTH passwords —
        # simulates a SQLAlchemy OperationalError that carries frame
        # locals from set_sqlcipher_rekey's traceback.
        exc = RuntimeError(
            f"sqlcipher rekey failed: old={_LEAKED_OLD_PASSWORD} "
            f"new={_LEAKED_PASSWORD} target={fake_db}"
        )
        manager.open_user_database = MagicMock(side_effect=exc)

        with loguru_caplog_full.at_level("DEBUG"):
            result = manager.change_password(
                "alice", _LEAKED_OLD_PASSWORD, _LEAKED_PASSWORD
            )

        assert result is False
        _assert_no_leak(
            loguru_caplog_full.text, _LEAKED_PASSWORD, "change_password new_pw"
        )
        _assert_no_leak(
            loguru_caplog_full.text,
            _LEAKED_OLD_PASSWORD,
            "change_password old_pw",
        )
        assert "Failed to change password" in loguru_caplog_full.text, (
            "test did not exercise the except branch — check the fixtures"
        )

    def test_open_user_database_outer_catch_does_not_leak(
        self, loguru_caplog_full, tmp_path
    ):
        """The outer ``except Exception`` in ``open_user_database`` (was
        line 688 in the issue's listing) fires when the engine fails to
        open. ``password`` is in scope and was passed into the engine
        creator closure via ``set_sqlcipher_key`` — an OperationalError
        traceback could leak it.
        """
        from local_deep_research.database.encrypted_db import DatabaseManager

        manager = DatabaseManager.__new__(DatabaseManager)
        # has_encryption=True drives the SQLCipher-engine branch, which
        # uses a custom creator with ``sqlite://`` (no event.listen).
        manager.has_encryption = True
        manager.connections = {}
        # __init__ is bypassed here; mirror its per-user init-lock dict so
        # open_user_database -> _get_init_lock has its backing store.
        manager._init_locks = {}
        manager._connections_lock = MagicMock()
        manager._connections_lock.__enter__ = lambda *a: None
        manager._connections_lock.__exit__ = lambda *a: None
        manager._pool_class = type("StubPool", (), {})
        manager._get_pool_kwargs = lambda: {}
        db_path = tmp_path / "alice.db"
        db_path.touch()
        manager._get_user_db_path = lambda u: db_path

        # Drive an exception from the SELECT 1 test connect that
        # follows engine creation.
        exc = RuntimeError(
            f"connect failed: dsn=sqlite:///x?key={_LEAKED_PASSWORD}"
        )

        with loguru_caplog_full.at_level("DEBUG"):
            with patch(
                "local_deep_research.database.encrypted_db.create_engine"
            ) as mock_create:
                with patch(
                    "local_deep_research.database.encrypted_db.has_per_database_salt",
                    return_value=True,
                ):
                    mock_engine = MagicMock()
                    mock_engine.connect.side_effect = exc
                    mock_create.return_value = mock_engine

                    result = manager.open_user_database(
                        "alice", _LEAKED_PASSWORD
                    )

        assert result is None
        _assert_no_leak(
            loguru_caplog_full.text,
            _LEAKED_PASSWORD,
            "open_user_database outer catch",
        )
        assert "Failed to open database" in loguru_caplog_full.text

    def test_open_user_database_migration_failure_sanitizes_typed_error(
        self, loguru_caplog_full, tmp_path
    ):
        """A migration failure re-raises as ``DatabaseInitializationError``.
        The typed error must carry a redacted message and no exception
        chain (``from None``, ADR-0003): callers (e.g.
        ``thread_local_session``) log it, and the broken chain plus
        redacted message ensure the original exception — and its frame
        locals holding the password — can't be rendered, defeating the
        redaction applied at the raise site.
        """
        from local_deep_research.database.encrypted_db import (
            DatabaseInitializationError,
            DatabaseManager,
        )

        manager = DatabaseManager.__new__(DatabaseManager)
        manager.has_encryption = True
        manager.connections = {}
        # __init__ is bypassed here; mirror its per-user init-lock dict so
        # open_user_database -> _get_init_lock has its backing store.
        manager._init_locks = {}
        manager._connections_lock = MagicMock()
        manager._connections_lock.__enter__ = lambda *a: None
        manager._connections_lock.__exit__ = lambda *a: None
        manager._pool_class = type("StubPool", (), {})
        manager._get_pool_kwargs = lambda: {}
        db_path = tmp_path / "alice.db"
        db_path.touch()
        manager._get_user_db_path = lambda u: db_path

        # The SELECT 1 connection check passes; the migration step then
        # raises with the password embedded (simulating an
        # OperationalError that carries frame locals).
        exc = RuntimeError(
            f"migration failed: dsn=sqlite:///x?key={_LEAKED_PASSWORD}"
        )

        with loguru_caplog_full.at_level("DEBUG"):
            with patch(
                "local_deep_research.database.encrypted_db.create_engine",
                return_value=MagicMock(),
            ):
                with patch(
                    "local_deep_research.database.encrypted_db.has_per_database_salt",
                    return_value=True,
                ):
                    with patch(
                        "local_deep_research.database.alembic_runner.needs_migration",
                        return_value=False,
                    ):
                        with patch(
                            "local_deep_research.database.initialize.initialize_database",
                            side_effect=exc,
                        ):
                            with pytest.raises(
                                DatabaseInitializationError
                            ) as excinfo:
                                manager.open_user_database(
                                    "alice", _LEAKED_PASSWORD
                                )

        # The typed error's own message must be redacted, and the chain
        # broken so a downstream logger.exception can't render init_err.
        _assert_no_leak(
            str(excinfo.value),
            _LEAKED_PASSWORD,
            "DatabaseInitializationError message",
        )
        assert excinfo.value.__cause__ is None
        assert excinfo.value.__suppress_context__ is True
        _assert_no_leak(
            loguru_caplog_full.text,
            _LEAKED_PASSWORD,
            "open_user_database migration catch",
        )
        assert "Database migration failed" in loguru_caplog_full.text


# ---------------------------------------------------------------------------
# web/queue/processor_v2.py
# ---------------------------------------------------------------------------


class TestProcessorV2PasswordLeakage:
    """``QueueProcessorV2._start_research_directly`` receives the user's
    password and forwards it to ``start_research_process``. A failure in
    the research startup path (or in the active-record DB write) whose
    exception embeds the password must not reach logs.
    """

    def test_start_research_directly_does_not_leak_on_research_failure(
        self, loguru_caplog_full
    ):
        from local_deep_research.web.queue.processor_v2 import (
            QueueProcessorV2,
        )

        processor = QueueProcessorV2.__new__(QueueProcessorV2)

        exc = RuntimeError(
            f"could not start research thread, "
            f"connection url=sqlite:///x.db?pwd={_LEAKED_PASSWORD}"
        )

        # The initial active-record create + status update succeeds, and
        # then start_research_process raises with the password in the
        # exception message.
        fake_session = MagicMock()
        fake_session.__enter__ = lambda s: s
        fake_session.__exit__ = lambda s, *a: None
        fake_session.query.return_value.filter_by.return_value.first.return_value = None

        with loguru_caplog_full.at_level("DEBUG"):
            with patch(
                "local_deep_research.web.queue.processor_v2.get_user_db_session",
                return_value=fake_session,
            ):
                with patch(
                    "local_deep_research.web.queue.processor_v2.UserQueueService"
                ):
                    with patch(
                        "local_deep_research.web.queue.processor_v2.start_research_process",
                        side_effect=exc,
                    ):
                        processor._start_research_directly(
                            username="alice",
                            research_id="r1",
                            password=_LEAKED_PASSWORD,
                            query="test",
                            mode="quick",
                        )

        _assert_no_leak(
            loguru_caplog_full.text,
            _LEAKED_PASSWORD,
            "_start_research_directly",
        )
        # Sanity: the failure branch ran.
        assert (
            "Failed to start research" in loguru_caplog_full.text
            or "Failed to create active research record"
            in loguru_caplog_full.text
        )

    def test_notify_research_queued_does_not_leak(self, loguru_caplog_full):
        """``notify_research_queued`` retrieves the session password and
        opens the user DB in its direct-execution branch, then falls back
        to a queue-status update. Both except handlers have ``password``
        in scope; an exception from either path must not leak it.
        """
        from local_deep_research.web.queue.processor_v2 import (
            QueueProcessorV2,
        )

        processor = QueueProcessorV2.__new__(QueueProcessorV2)

        exc = RuntimeError(f"open failed: key={_LEAKED_PASSWORD} db=alice.db")

        with loguru_caplog_full.at_level("DEBUG"):
            with (
                patch(
                    "local_deep_research.web.queue.processor_v2.session_password_store"
                ) as mock_store,
                patch(
                    "local_deep_research.web.queue.processor_v2.db_manager"
                ) as mock_db,
                patch(
                    "local_deep_research.web.queue.processor_v2.get_user_db_session",
                    side_effect=exc,
                ),
            ):
                mock_store.get_session_password.return_value = _LEAKED_PASSWORD
                mock_db.open_user_database.side_effect = exc

                processor.notify_research_queued(
                    "alice", "r1", session_id="s1", query="q", mode="quick"
                )

        _assert_no_leak(
            loguru_caplog_full.text,
            _LEAKED_PASSWORD,
            "notify_research_queued",
        )
        # Sanity: both except branches ran (direct-execution error, then
        # the queue-status fallback error).
        assert "Error in direct execution" in loguru_caplog_full.text
        assert "Failed to update queue status" in loguru_caplog_full.text

    def test_process_user_queue_does_not_leak(self, loguru_caplog_full):
        """``_process_user_queue`` retrieves the session password before
        its try block and passes it into ``open_user_database`` /
        ``get_user_db_session``. The catch-all must redact it.
        """
        from local_deep_research.web.queue.processor_v2 import (
            QueueProcessorV2,
        )

        processor = QueueProcessorV2.__new__(QueueProcessorV2)

        exc = RuntimeError(f"sqlcipher key rejected: {_LEAKED_PASSWORD}")

        with loguru_caplog_full.at_level("DEBUG"):
            with (
                patch(
                    "local_deep_research.web.queue.processor_v2.session_password_store"
                ) as mock_store,
                patch(
                    "local_deep_research.web.queue.processor_v2.db_manager"
                ) as mock_db,
            ):
                mock_store.get_session_password.return_value = _LEAKED_PASSWORD
                mock_db.open_user_database.side_effect = exc

                result = processor._process_user_queue("alice", "s1")

        assert result is False
        _assert_no_leak(
            loguru_caplog_full.text,
            _LEAKED_PASSWORD,
            "_process_user_queue",
        )
        assert "Error processing queue for user" in loguru_caplog_full.text

    def test_notify_research_completed_does_not_leak(self, loguru_caplog_full):
        """``notify_research_completed`` has ``user_password`` as a
        parameter and passes it into ``get_user_db_session``.
        """
        from local_deep_research.web.queue.processor_v2 import (
            QueueProcessorV2,
        )

        processor = QueueProcessorV2.__new__(QueueProcessorV2)

        exc = RuntimeError(
            f"session open failed: dsn=sqlite:///x?key={_LEAKED_PASSWORD}"
        )

        with loguru_caplog_full.at_level("DEBUG"):
            with (
                patch(
                    "local_deep_research.web.queue.processor_v2.get_user_db_session",
                    side_effect=exc,
                ),
                patch(
                    "local_deep_research.research_library.search.services.research_history_indexer.auto_convert_research"
                ),
            ):
                processor.notify_research_completed(
                    "alice", "r1", user_password=_LEAKED_PASSWORD
                )

        _assert_no_leak(
            loguru_caplog_full.text,
            _LEAKED_PASSWORD,
            "notify_research_completed",
        )
        assert "Failed to update completion status" in loguru_caplog_full.text

    def test_notify_research_failed_does_not_leak(self, loguru_caplog_full):
        """``notify_research_failed`` — same contract as the completed
        notification.
        """
        from local_deep_research.web.queue.processor_v2 import (
            QueueProcessorV2,
        )

        processor = QueueProcessorV2.__new__(QueueProcessorV2)

        exc = RuntimeError(
            f"session open failed: dsn=sqlite:///x?key={_LEAKED_PASSWORD}"
        )

        with loguru_caplog_full.at_level("DEBUG"):
            with patch(
                "local_deep_research.web.queue.processor_v2.get_user_db_session",
                side_effect=exc,
            ):
                processor.notify_research_failed(
                    "alice",
                    "r1",
                    error_message="boom",
                    user_password=_LEAKED_PASSWORD,
                )

        _assert_no_leak(
            loguru_caplog_full.text,
            _LEAKED_PASSWORD,
            "notify_research_failed",
        )
        assert "Failed to update failure status" in loguru_caplog_full.text

    def test_start_queued_researches_spawn_failure_does_not_leak(
        self, loguru_caplog_full
    ):
        """The spawn-failure handler in ``_start_queued_researches`` has
        ``password`` as a parameter. A spawn exception whose message
        embeds it must be redacted — and the rollback-failure debug log
        in the same handler must not re-render the traceback.
        """
        import threading

        from local_deep_research.web.queue.processor_v2 import (
            QueueProcessorV2,
        )

        processor = QueueProcessorV2.__new__(QueueProcessorV2)
        processor._spawn_retry_counts = {}
        processor._spawn_retry_counts_lock = threading.Lock()
        processor._reclaim_stranded_queue_rows = lambda *a: None

        exc = RuntimeError(
            f"spawn failed: thread env carried pwd={_LEAKED_PASSWORD}"
        )
        processor._start_research = MagicMock(side_effect=exc)

        queued_item = MagicMock()
        queued_item.research_id = "r1"

        db_session = MagicMock()
        query_chain = db_session.query.return_value.filter_by.return_value
        query_chain.order_by.return_value.limit.return_value.all.return_value = [
            queued_item
        ]
        query_chain.update.return_value = 1  # claim succeeds
        # Rollback after the spawn failure raises too, with the password
        # embedded — exercises the redacted debug path.
        db_session.rollback.side_effect = RuntimeError(
            f"rollback failed: {_LEAKED_PASSWORD}"
        )

        with loguru_caplog_full.at_level("DEBUG"):
            processor._start_queued_researches(
                db_session,
                MagicMock(),  # queue_service
                "alice",
                _LEAKED_PASSWORD,
                available_slots=1,
            )

        _assert_no_leak(
            loguru_caplog_full.text,
            _LEAKED_PASSWORD,
            "_start_queued_researches spawn failure",
        )
        assert "Error starting queued research" in loguru_caplog_full.text

    def test_process_user_request_does_not_leak(self, loguru_caplog_full):
        """``process_user_request`` retrieves the session password inside
        its try block (hence the ``password = None`` pre-declaration) and
        opens the user DB. The catch-all must redact it.
        """
        import threading

        from local_deep_research.web.queue.processor_v2 import (
            QueueProcessorV2,
        )

        processor = QueueProcessorV2.__new__(QueueProcessorV2)
        processor._users_to_check = set()
        processor._users_lock = threading.Lock()

        exc = RuntimeError(f"open failed, key in dsn: {_LEAKED_PASSWORD}")

        with loguru_caplog_full.at_level("DEBUG"):
            with (
                patch(
                    "local_deep_research.web.queue.processor_v2.session_password_store"
                ) as mock_store,
                patch(
                    "local_deep_research.web.queue.processor_v2.db_manager"
                ) as mock_db,
            ):
                mock_store.get_session_password.return_value = _LEAKED_PASSWORD
                mock_db.open_user_database.side_effect = exc

                result = processor.process_user_request("alice", "s1")

        assert result == 0
        _assert_no_leak(
            loguru_caplog_full.text,
            _LEAKED_PASSWORD,
            "process_user_request",
        )
        assert "Error in process_user_request" in loguru_caplog_full.text


# ---------------------------------------------------------------------------
# scheduler/background.py
# ---------------------------------------------------------------------------


class TestSchedulerBackgroundPasswordLeakage:
    """The scheduler retrieves user passwords from its credential store
    and passes them into encrypted-DB session contexts. Every
    ``except Exception`` site in those flows must scrub the password
    from str(e) before logging.
    """

    @pytest.fixture
    def scheduler(self):
        """Build a ``BackgroundJobScheduler`` with a populated credential
        store but no actual APScheduler running — enough for the except
        handlers to be exercised in isolation.
        """
        from local_deep_research.scheduler.background import (
            BackgroundJobScheduler,
            SchedulerCredentialStore,
        )

        sched = BackgroundJobScheduler.__new__(BackgroundJobScheduler)
        sched._credential_store = SchedulerCredentialStore(ttl_hours=1)
        sched._credential_store.store("alice", _LEAKED_PASSWORD)
        sched.user_sessions = {
            "alice": {"last_activity": None, "scheduled_jobs": set()}
        }
        sched.lock = MagicMock()
        sched.lock.__enter__ = lambda *a: None
        sched.lock.__exit__ = lambda *a: None
        sched.scheduler = MagicMock()
        sched.is_running = True
        sched.config = {}
        return sched

    def test_trigger_subscription_research_sync_does_not_leak(
        self, loguru_caplog_full, scheduler
    ):
        """``_trigger_subscription_research_sync`` retrieves the
        password from the credential store and passes it through to
        ``get_user_db_session`` and ``quick_summary``. A SQLAlchemy
        / requests exception from any of those paths must not surface
        the password.
        """
        exc = RuntimeError(f"db session open failed: pwd={_LEAKED_PASSWORD}")

        with loguru_caplog_full.at_level("DEBUG"):
            with patch(
                "local_deep_research.database.session_context.get_user_db_session",
                side_effect=exc,
            ):
                scheduler._trigger_subscription_research_sync(
                    "alice",
                    {"id": 42, "name": "test sub", "query": "test"},
                )

        _assert_no_leak(
            loguru_caplog_full.text,
            _LEAKED_PASSWORD,
            "_trigger_subscription_research_sync",
        )
        assert "Error triggering research" in loguru_caplog_full.text

    def test_schedule_user_subscriptions_does_not_leak(
        self, loguru_caplog_full, scheduler
    ):
        """``_schedule_user_subscriptions`` retrieves the password and
        opens an encrypted DB session. Failure inside that block must
        not leak the password.
        """
        exc = RuntimeError(
            f"NewsSubscription query failed: dsn=sqlite:///x?pwd={_LEAKED_PASSWORD}"
        )

        with loguru_caplog_full.at_level("DEBUG"):
            with patch(
                "local_deep_research.database.session_context.get_user_db_session",
                side_effect=exc,
            ):
                # _schedule_user_subscriptions also calls
                # _schedule_document_processing at the end — stub it so
                # we isolate the failure to the subscription branch.
                scheduler._schedule_document_processing = lambda u: None
                scheduler._schedule_user_subscriptions("alice")

        _assert_no_leak(
            loguru_caplog_full.text,
            _LEAKED_PASSWORD,
            "_schedule_user_subscriptions",
        )
        assert "Error scheduling subscriptions" in loguru_caplog_full.text

    def test_check_user_overdue_subscriptions_does_not_leak(
        self, loguru_caplog_full, scheduler
    ):
        """``_check_user_overdue_subscriptions`` retrieves the password
        and opens an encrypted DB session. Same contract.
        """
        exc = RuntimeError(
            f"overdue query failed, pwd={_LEAKED_PASSWORD} in dsn"
        )

        with loguru_caplog_full.at_level("DEBUG"):
            with patch(
                "local_deep_research.database.session_context.get_user_db_session",
                side_effect=exc,
            ):
                scheduler._check_user_overdue_subscriptions("alice")

        _assert_no_leak(
            loguru_caplog_full.text,
            _LEAKED_PASSWORD,
            "_check_user_overdue_subscriptions",
        )
        assert "Error checking overdue subscriptions" in loguru_caplog_full.text

    def test_check_subscription_does_not_leak(
        self, loguru_caplog_full, scheduler
    ):
        """``_check_subscription`` retrieves the password from the
        credential store and opens an encrypted DB session. A failure
        from that path must not surface the password.
        """
        exc = RuntimeError(
            f"subscription refresh failed: dsn=sqlite:///x?pwd={_LEAKED_PASSWORD}"
        )

        with loguru_caplog_full.at_level("DEBUG"):
            with patch(
                "local_deep_research.database.session_context.get_user_db_session",
                side_effect=exc,
            ):
                scheduler._check_subscription("alice", 42)

        _assert_no_leak(
            loguru_caplog_full.text,
            _LEAKED_PASSWORD,
            "_check_subscription",
        )
        assert "Error checking subscription 42" in loguru_caplog_full.text

    def test_get_document_scheduler_settings_does_not_leak(
        self, loguru_caplog_full, scheduler
    ):
        """``_get_document_scheduler_settings`` retrieves the password
        before its try block and passes it into ``get_user_db_session``.
        On failure it must log redacted and fall back to defaults.
        """
        import threading

        scheduler._settings_cache = {}
        scheduler._settings_cache_lock = threading.Lock()

        exc = RuntimeError(
            f"settings fetch failed: key={_LEAKED_PASSWORD} db=alice"
        )

        with loguru_caplog_full.at_level("DEBUG"):
            with patch(
                "local_deep_research.database.session_context.get_user_db_session",
                side_effect=exc,
            ):
                settings = scheduler._get_document_scheduler_settings("alice")

        # Falls back to defaults rather than raising.
        assert settings is not None
        _assert_no_leak(
            loguru_caplog_full.text,
            _LEAKED_PASSWORD,
            "_get_document_scheduler_settings",
        )
        assert "Error fetching settings" in loguru_caplog_full.text

    def test_store_research_result_does_not_leak(
        self, loguru_caplog_full, scheduler
    ):
        """``_store_research_result`` receives the password as a
        function parameter and passes it into ``get_user_db_session``.
        A failure anywhere inside the storage block must not surface
        the password.
        """
        exc = RuntimeError(
            f"history insert failed: dsn=sqlite:///x?pwd={_LEAKED_PASSWORD}"
        )

        with loguru_caplog_full.at_level("DEBUG"):
            with patch(
                "local_deep_research.database.session_context.get_user_db_session",
                side_effect=exc,
            ):
                scheduler._store_research_result(
                    username="alice",
                    password=_LEAKED_PASSWORD,
                    research_id="r1",
                    subscription_id=42,
                    result={},
                    subscription={},
                )

        _assert_no_leak(
            loguru_caplog_full.text,
            _LEAKED_PASSWORD,
            "_store_research_result",
        )
        assert "Error storing research result" in loguru_caplog_full.text

    def test_process_user_documents_snapshot_failure_does_not_leak(
        self, loguru_caplog_full, scheduler
    ):
        """The settings-snapshot handler inside ``_process_user_documents``
        runs nested inside ``get_user_db_session(username, password)`` —
        the password is live in the frame. A failure building the
        snapshot must not surface it.
        """
        exc = RuntimeError(
            f"snapshot build failed: dsn=sqlite:///x?pwd={_LEAKED_PASSWORD}"
        )

        # Document-scheduler settings: one processing flag enabled so
        # the function reaches the snapshot block instead of returning
        # early; no last_run so the date filter is skipped.
        doc_settings = MagicMock(
            download_pdfs=True,
            extract_text=False,
            generate_rag=False,
            last_run=None,
        )
        scheduler._get_document_scheduler_settings = lambda u: doc_settings
        scheduler._arm_egress_backstop = lambda sm, u: None

        # One fake completed research session so the snapshot block is
        # reached (empty result returns before it).
        fake_research = MagicMock(id="r1", title="t", completed_at=None)
        fake_db = MagicMock()
        fake_db.query.return_value.filter.return_value.order_by.return_value.limit.return_value.all.return_value = [
            fake_research
        ]
        fake_session = MagicMock()
        fake_session.__enter__ = lambda s: fake_db
        fake_session.__exit__ = lambda s, *a: None

        with loguru_caplog_full.at_level("DEBUG"):
            with patch(
                "local_deep_research.database.session_context.get_user_db_session",
                return_value=fake_session,
            ):
                with patch(
                    "local_deep_research.settings.manager.SettingsManager"
                ) as mock_sm:
                    mock_sm.return_value.get_settings_snapshot.side_effect = exc
                    with patch(
                        "local_deep_research.research_library.services.download_service.DownloadService"
                    ):
                        scheduler._process_user_documents("alice")

        _assert_no_leak(
            loguru_caplog_full.text,
            _LEAKED_PASSWORD,
            "_process_user_documents snapshot handler",
        )
        assert "Could not build settings snapshot" in loguru_caplog_full.text