"""
Regression tests for issue #3827 — Download Manager fails with
``UNIQUE constraint failed: documents.document_hash`` after 20–30 PDFs.

The bug had two parts:

1. ``_save_text_with_db`` and ``_download_pdf`` unconditionally overwrote
   ``Document.document_hash`` on every existing-document update. The PDF-bytes
   hash (essentially collision-free) was clobbered with a text-content hash
   (collision-prone). After enough downloads, two PDFs that extracted to
   identical text produced the same hash and the second commit raised
   ``IntegrityError``.

2. No rollback ran after the IntegrityError, so the shared thread-local
   session stayed in ``PendingRollbackError`` state and every subsequent
   operation cascaded.

The fix preserves PR #2590's retry-from-FAILED intent (still replace the
placeholder ``failed:url:resource_id`` hash with a real content hash) while
leaving stable hashes alone on normal updates, adds a lookup-before-insert in
the new-Document branch, and rolls back on the inner exception path.

These tests use a real in-memory SQLite session so the UNIQUE constraint
behaviour is exercised end-to-end.
"""

import hashlib
import uuid
from unittest.mock import MagicMock, patch

import pytest
from sqlalchemy import create_engine
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import sessionmaker

from local_deep_research.database.models import Base
from local_deep_research.database.models.library import (
    Collection,
    Document,
    DocumentStatus,
    SourceType,
)
from local_deep_research.database.models.research import (
    ResearchHistory,
    ResearchResource,
)
from local_deep_research.research_library.services.download_service import (
    DownloadService,
)


MODULE = "local_deep_research.research_library.services.download_service"


@pytest.fixture
def session():
    """Real in-memory SQLite session — exercises actual UNIQUE constraints."""
    engine = create_engine("sqlite:///:memory:")
    Base.metadata.create_all(engine)
    Session = sessionmaker(bind=engine)
    session = Session()
    yield session
    session.close()
    engine.dispose()


@pytest.fixture
def source_type(session):
    src = SourceType(
        id=str(uuid.uuid4()),
        name="research_download",
        display_name="Research Download",
    )
    session.add(src)
    session.commit()
    return src


@pytest.fixture
def library_collection(session):
    coll = Collection(
        id=str(uuid.uuid4()),
        name="Library",
        is_default=True,
        collection_type="default_library",
    )
    session.add(coll)
    session.commit()
    return coll


@pytest.fixture
def research(session):
    """ResearchHistory row required by ResearchResource FK."""
    rh = ResearchHistory(
        id=str(uuid.uuid4()),
        query="test query",
        mode="quick",
        status="completed",
        created_at="2026-05-09T00:00:00",
    )
    session.add(rh)
    session.commit()
    return rh


@pytest.fixture
def make_resource(session, research):
    def _make(url):
        r = ResearchResource(
            research_id=research.id,
            title="Test Paper",
            url=url,
            source_type="academic",
            created_at="2026-05-09T00:00:00",
        )
        session.add(r)
        session.commit()
        return r

    return _make


@pytest.fixture
def svc():
    """DownloadService stub — only the methods we test directly are exercised."""
    with patch.object(DownloadService, "__init__", lambda self, *a, **kw: None):
        s = DownloadService.__new__(DownloadService)
        s.username = "test_user"
        s.password = "test_pass"
        s._closed = False
        return s


# ---------------------------------------------------------------------------
# Change 1 / site 2: _save_text_with_db must NOT overwrite hash on
# normal (status=COMPLETED) updates — this is the primary #3827 bug.
# ---------------------------------------------------------------------------


def test_completed_doc_hash_is_not_overwritten_by_text_extraction(
    session, source_type, make_resource, svc
):
    """Two PDFs with different bytes whose extractors return identical text
    used to collide on the text-hash overwrite. With the fix, each Document
    keeps its PDF-bytes hash and no UNIQUE collision occurs.
    """
    res_a = make_resource("https://arxiv.org/abs/2401.A")
    res_b = make_resource("https://arxiv.org/abs/2401.B")

    pdf_a_hash = hashlib.sha256(b"PDF-A bytes").hexdigest()
    pdf_b_hash = hashlib.sha256(b"PDF-B bytes").hexdigest()
    identical_text = "Identical extracted body text from both PDFs."

    # Mirror what _download_pdf would have created at line 664.
    doc_a = Document(
        id=str(uuid.uuid4()),
        source_type_id=source_type.id,
        resource_id=res_a.id,
        research_id=res_a.research_id,
        document_hash=pdf_a_hash,
        original_url=res_a.url,
        file_size=11,
        file_type="pdf",
        title=res_a.title,
        status=DocumentStatus.COMPLETED,
    )
    doc_b = Document(
        id=str(uuid.uuid4()),
        source_type_id=source_type.id,
        resource_id=res_b.id,
        research_id=res_b.research_id,
        document_hash=pdf_b_hash,
        original_url=res_b.url,
        file_size=11,
        file_type="pdf",
        title=res_b.title,
        status=DocumentStatus.COMPLETED,
    )
    session.add_all([doc_a, doc_b])
    session.commit()

    with patch(f"{MODULE}.get_document_for_resource") as gdr:
        gdr.side_effect = [doc_a, doc_b]
        svc._save_text_with_db(
            res_a,
            identical_text,
            session,
            extraction_method="pdf_extraction",
            extraction_source="pdfplumber",
            pdf_document_id=doc_a.id,
        )
        svc._save_text_with_db(
            res_b,
            identical_text,
            session,
            extraction_method="pdf_extraction",
            extraction_source="pdfplumber",
            pdf_document_id=doc_b.id,
        )
        # The second commit used to raise IntegrityError. With the fix, both
        # commits succeed and the hashes are unchanged.
        session.commit()

    assert doc_a.document_hash == pdf_a_hash
    assert doc_b.document_hash == pdf_b_hash
    assert doc_a.text_content == identical_text
    assert doc_b.text_content == identical_text


# ---------------------------------------------------------------------------
# Change 1 retry preservation: PR #2590's intent is preserved — a Document
# whose status was FAILED still gets its placeholder hash replaced when
# real text becomes available.
# ---------------------------------------------------------------------------


def test_failed_doc_hash_is_replaced_with_text_hash_on_retry(
    session, source_type, make_resource, svc
):
    res = make_resource("https://arxiv.org/abs/2401.C")
    placeholder = hashlib.sha256(
        f"failed:{res.url}:{res.id}".encode()
    ).hexdigest()

    failed_doc = Document(
        id=str(uuid.uuid4()),
        source_type_id=source_type.id,
        resource_id=res.id,
        research_id=res.research_id,
        document_hash=placeholder,
        original_url=res.url,
        file_size=0,
        file_type="unknown",
        title=res.title,
        status=DocumentStatus.FAILED,
    )
    session.add(failed_doc)
    session.commit()

    extracted_text = "Real text extracted on retry."
    expected_hash = hashlib.sha256(extracted_text.encode()).hexdigest()

    with patch(f"{MODULE}.get_document_for_resource", return_value=failed_doc):
        svc._save_text_with_db(
            res,
            extracted_text,
            session,
            extraction_method="native_api",
            extraction_source="arxiv",
        )
        session.commit()

    assert failed_doc.document_hash == expected_hash
    assert failed_doc.status == DocumentStatus.COMPLETED


# ---------------------------------------------------------------------------
# Change 2: lookup-before-insert in the new-text-only-Document branch
# prevents UNIQUE collisions when two resources have no PDF and produce
# identical extracted text via the API path.
# ---------------------------------------------------------------------------


def test_text_only_extraction_dedups_on_existing_content_hash(
    session, source_type, library_collection, make_resource, svc
):
    res_a = make_resource("https://example.com/api/A")
    res_b = make_resource("https://example.com/api/B")

    text = "Same text from two different API endpoints."
    text_hash = hashlib.sha256(text.encode()).hexdigest()

    with (
        patch(f"{MODULE}.get_document_for_resource", return_value=None),
        patch(f"{MODULE}.get_source_type_id", return_value=source_type.id),
    ):
        svc._save_text_with_db(
            res_a,
            text,
            session,
            extraction_method="native_api",
            extraction_source="arxiv",
        )
        session.commit()
        svc._save_text_with_db(
            res_b,
            text,
            session,
            extraction_method="native_api",
            extraction_source="arxiv",
        )
        session.commit()

    docs = session.query(Document).filter_by(document_hash=text_hash).all()
    assert len(docs) == 1, (
        f"Dedup should have produced exactly one Document, got {len(docs)}"
    )
    canonical = docs[0]

    session.refresh(res_b)
    assert res_b.document_id == canonical.id, (
        "Second resource should be linked to the canonical Document."
    )


# ---------------------------------------------------------------------------
# Change 5: _save_text_with_db's outer except calls session.rollback() before
# re-raising, so the caller's loop sees a clean session on the next iteration.
# ---------------------------------------------------------------------------


def test_save_text_with_db_rolls_back_session_on_exception(svc):
    """When _save_text_with_db's inner work raises, the session must be
    rolled back before the exception propagates so the next caller's
    iteration doesn't trip over PendingRollbackError.
    """
    session = MagicMock()
    resource = MagicMock()

    with patch(
        f"{MODULE}.get_document_for_resource",
        side_effect=RuntimeError("simulated flush failure"),
    ):
        with pytest.raises(RuntimeError, match="simulated flush failure"):
            svc._save_text_with_db(
                resource,
                "some text",
                session,
                extraction_method="native_api",
                extraction_source="arxiv",
            )

    session.rollback.assert_called_once()


# ---------------------------------------------------------------------------
# Change 1 / site 1: _download_pdf existing-doc branch should leave the hash
# alone for COMPLETED docs and replace the placeholder for FAILED docs.
# We exercise the branch directly via the same code path used by _download_pdf
# (ASCII-test of the conditional is the cleanest assertion here).
# ---------------------------------------------------------------------------


def test_download_pdf_existing_completed_doc_hash_is_stable(
    session, source_type, make_resource
):
    """Mirror line 614-619's behaviour: a COMPLETED doc must not have its
    hash overwritten on re-download. We construct the in-memory state and
    simulate the conditional update directly.
    """
    res = make_resource("https://arxiv.org/abs/2401.D")
    original_hash = hashlib.sha256(b"original PDF").hexdigest()
    doc = Document(
        id=str(uuid.uuid4()),
        source_type_id=source_type.id,
        resource_id=res.id,
        research_id=res.research_id,
        document_hash=original_hash,
        original_url=res.url,
        file_size=12,
        file_type="pdf",
        title=res.title,
        status=DocumentStatus.COMPLETED,
    )
    session.add(doc)
    session.commit()

    new_pdf_hash = hashlib.sha256(b"different PDF bytes").hexdigest()

    # Mirrored conditional from download_service.py:614-622
    was_failed = doc.status == DocumentStatus.FAILED
    if was_failed:
        doc.document_hash = new_pdf_hash
    doc.status = DocumentStatus.COMPLETED
    session.commit()

    assert doc.document_hash == original_hash, (
        "Completed doc's hash must remain stable; only FAILED retries replace it."
    )


def test_download_pdf_existing_failed_doc_hash_is_replaced(
    session, source_type, make_resource
):
    res = make_resource("https://arxiv.org/abs/2401.E")
    placeholder = hashlib.sha256(
        f"failed:{res.url}:{res.id}".encode()
    ).hexdigest()
    doc = Document(
        id=str(uuid.uuid4()),
        source_type_id=source_type.id,
        resource_id=res.id,
        research_id=res.research_id,
        document_hash=placeholder,
        original_url=res.url,
        file_size=0,
        file_type="unknown",
        title=res.title,
        status=DocumentStatus.FAILED,
    )
    session.add(doc)
    session.commit()

    new_pdf_hash = hashlib.sha256(
        b"PDF bytes from successful retry"
    ).hexdigest()

    was_failed = doc.status == DocumentStatus.FAILED
    if was_failed:
        doc.document_hash = new_pdf_hash
    doc.status = DocumentStatus.COMPLETED
    session.commit()

    assert doc.document_hash == new_pdf_hash
    assert doc.status == DocumentStatus.COMPLETED


# ---------------------------------------------------------------------------
# Sanity check: confirm that WITHOUT the fix, two identical-text updates DO
# violate the unique constraint. This guards against a regression that
# silently makes test 1 trivially pass (e.g. a schema change that drops the
# constraint).
# ---------------------------------------------------------------------------


def test_unique_constraint_still_enforced_at_db_level(
    session, source_type, make_resource
):
    res_a = make_resource("https://example.com/a")
    res_b = make_resource("https://example.com/b")

    shared_hash = hashlib.sha256(b"shared content").hexdigest()
    doc_a = Document(
        id=str(uuid.uuid4()),
        source_type_id=source_type.id,
        resource_id=res_a.id,
        research_id=res_a.research_id,
        document_hash=shared_hash,
        original_url=res_a.url,
        file_size=14,
        file_type="text",
        title="A",
        status=DocumentStatus.COMPLETED,
    )
    doc_b = Document(
        id=str(uuid.uuid4()),
        source_type_id=source_type.id,
        resource_id=res_b.id,
        research_id=res_b.research_id,
        document_hash=shared_hash,
        original_url=res_b.url,
        file_size=14,
        file_type="text",
        title="B",
        status=DocumentStatus.COMPLETED,
    )
    session.add_all([doc_a, doc_b])
    with pytest.raises(IntegrityError):
        session.commit()
    session.rollback()


# ---------------------------------------------------------------------------
# Stress test: replays the exact bug scenario from the issue report — many
# PDFs whose extractors collapse to identical text. Without the fix this
# raised IntegrityError after a handful; with the fix the entire batch
# completes.
# ---------------------------------------------------------------------------


def test_25_pdfs_with_identical_extracted_text_all_succeed(
    session, source_type, make_resource, svc
):
    """The original report describes failure 'after 20 to 30 PDFs'. This test
    drives 25 distinct PDF Documents whose extractors all return the same
    text — the worst-case happy-path scenario for the old code, where every
    `_save_text_with_db` call would clobber the hash to the same text-hash
    and the second commit onwards would raise UNIQUE constraint failures.
    """
    n = 25
    docs = []
    for i in range(n):
        res = make_resource(f"https://arxiv.org/abs/2401.{i:04d}")
        # Each PDF has unique bytes (and therefore a unique PDF-bytes hash).
        pdf_hash = hashlib.sha256(f"PDF-bytes-{i}".encode()).hexdigest()
        doc = Document(
            id=str(uuid.uuid4()),
            source_type_id=source_type.id,
            resource_id=res.id,
            research_id=res.research_id,
            document_hash=pdf_hash,
            original_url=res.url,
            file_size=64,
            file_type="pdf",
            title=res.title,
            status=DocumentStatus.COMPLETED,
        )
        session.add(doc)
        docs.append((res, doc, pdf_hash))
    session.commit()

    identical_text = "Same extracted text from all 25 PDFs."

    # Iterate exactly as the production loop does — each call goes through
    # the existing-Document branch (line 1584) of _save_text_with_db.
    with patch(f"{MODULE}.get_document_for_resource") as gdr:
        gdr.side_effect = [d for (_, d, _) in docs]
        for res, doc, _ in docs:
            svc._save_text_with_db(
                res,
                identical_text,
                session,
                extraction_method="pdf_extraction",
                extraction_source="pdfplumber",
                pdf_document_id=doc.id,
            )
        # Single commit at the end so any pending IntegrityError surfaces here.
        session.commit()

    # Every Document kept its unique PDF-bytes hash; none collapsed to the
    # text-hash that would have collided.
    for _, doc, pdf_hash in docs:
        assert doc.document_hash == pdf_hash
        assert doc.text_content == identical_text

    # And of course no duplicates exist in the table.
    hashes = {d.document_hash for (_, d, _) in docs}
    assert len(hashes) == n


# ---------------------------------------------------------------------------
# Status-matrix coverage: confirms the conditional fires ONLY for
# DocumentStatus.FAILED. Future-proofs against someone broadening the
# trigger and silently re-introducing the bug.
# ---------------------------------------------------------------------------


@pytest.mark.parametrize(
    "prior_status,expected_overwrite",
    [
        (DocumentStatus.FAILED, True),
        (DocumentStatus.COMPLETED, False),
        (DocumentStatus.PENDING, False),
        (DocumentStatus.PROCESSING, False),
    ],
)
def test_save_text_with_db_overwrite_is_gated_on_failed_only(
    session, source_type, make_resource, svc, prior_status, expected_overwrite
):
    res = make_resource(f"https://example.com/status/{prior_status.value}")
    original_hash = hashlib.sha256(
        f"original-{prior_status.value}".encode()
    ).hexdigest()
    doc = Document(
        id=str(uuid.uuid4()),
        source_type_id=source_type.id,
        resource_id=res.id,
        research_id=res.research_id,
        document_hash=original_hash,
        original_url=res.url,
        file_size=10,
        file_type="pdf",
        title=res.title,
        status=prior_status,
    )
    session.add(doc)
    session.commit()

    text = f"Extracted text for {prior_status.value}"
    text_hash = hashlib.sha256(text.encode()).hexdigest()

    with patch(f"{MODULE}.get_document_for_resource", return_value=doc):
        svc._save_text_with_db(
            res,
            text,
            session,
            extraction_method="pdf_extraction",
            extraction_source="pdfplumber",
        )
        session.commit()

    if expected_overwrite:
        assert doc.document_hash == text_hash, (
            f"Hash should have been replaced for prior status {prior_status}"
        )
    else:
        assert doc.document_hash == original_hash, (
            f"Hash must remain stable for prior status {prior_status}; "
            f"only FAILED triggers replacement."
        )
    # In every case the doc ends up COMPLETED.
    assert doc.status == DocumentStatus.COMPLETED


# ---------------------------------------------------------------------------
# Change 4: the SSE bulk-download generator's per-item except calls
# session.rollback() at the TOP of the block, so the next iteration's
# pre-loop session.query (line 843) doesn't trip on a poisoned session.
# This is structural — we drive the `download_bulk` route end-to-end with
# Flask's test client.
# ---------------------------------------------------------------------------


def test_download_bulk_rolls_back_session_when_download_raises():
    """Drive `download_bulk` with two queue items; the first triggers
    IntegrityError. Verify `session.rollback()` is called on the OUTER
    request session before the next iteration would re-query.
    """
    from contextlib import contextmanager
    from flask import Flask

    from local_deep_research.research_library.routes import (
        library_routes as routes_mod,
    )

    # Two queue items so we can prove the second iteration sees a clean
    # session.
    q_item_a = MagicMock()
    q_item_a.resource_id = 101
    q_item_a.collection_id = None
    q_item_b = MagicMock()
    q_item_b.resource_id = 102
    q_item_b.collection_id = None

    resource_a = MagicMock()
    resource_a.id = 101
    resource_a.url = "https://example.com/a.pdf"
    resource_a.title = "A"
    resource_a.research_id = "r1"
    resource_b = MagicMock()
    resource_b.id = 102
    resource_b.url = "https://example.com/b.pdf"
    resource_b.title = "B"
    resource_b.research_id = "r1"

    outer_session = MagicMock()
    # Track session.query side effects in order:
    #  - count() for total pending (returns 2)
    #  - .all() for queue_items list (returns [a, b])
    #  - .get(101) for resource A (line 843 first iteration)
    #  - .get(102) for resource B (line 843 second iteration — must succeed
    #    after rollback)
    count_q = MagicMock()
    count_q.filter_by.return_value = count_q
    count_q.count.return_value = 2
    items_q = MagicMock()
    items_q.filter_by.return_value = items_q
    items_q.all.return_value = [q_item_a, q_item_b]
    resource_q = MagicMock()
    resource_q.get.side_effect = [resource_a, resource_b]
    outer_session.query.side_effect = [
        count_q,
        items_q,
        resource_q,
        resource_q,
    ]

    @contextmanager
    def fake_db_session(*a, **kw):
        yield outer_session

    download_service_mock = MagicMock()
    download_service_mock.__enter__ = MagicMock(
        return_value=download_service_mock
    )
    download_service_mock.__exit__ = MagicMock(return_value=False)
    # First call raises an IntegrityError-like; second succeeds.
    download_service_mock.download_resource.side_effect = [
        IntegrityError("INSERT", {}, Exception("UNIQUE constraint failed")),
        (True, None),
    ]

    app = Flask(__name__)
    app.config["SECRET_KEY"] = "test"
    app.config["TESTING"] = True
    app.register_blueprint(routes_mod.library_bp)

    with (
        patch.object(
            routes_mod, "get_user_db_session", side_effect=fake_db_session
        ),
        patch.object(
            routes_mod,
            "get_authenticated_user_password",
            return_value="pw",
        ),
        patch.object(
            routes_mod, "DownloadService", return_value=download_service_mock
        ),
        patch(
            "local_deep_research.web.auth.decorators.db_manager",
            MagicMock(
                is_user_connected=MagicMock(return_value=True),
                connections={"testuser": True},
                has_encryption=False,
            ),
        ),
    ):
        with app.test_client() as client:
            with client.session_transaction() as sess:
                sess["username"] = "testuser"
                sess["session_id"] = "sid"
            resp = client.post(
                "/library/api/download-bulk",
                json={"research_ids": ["r1"], "mode": "pdf"},
            )
            # The SSE generator is lazy — it only runs as the body is
            # consumed. Force consumption so the per-item loop executes
            # and the rollback paths fire.
            body = resp.get_data(as_text=True)

    assert resp.status_code == 200, body
    # The first item raised IntegrityError; the rollback must have fired
    # at the top of the per-item except handler so the loop could query
    # resource B for the second iteration.
    assert outer_session.rollback.called, (
        "session.rollback() must be called when download_resource raises, "
        "before the next iteration's session.query at line 843."
    )
    # Both queue items were attempted (not just the first — confirms the
    # loop continued on a clean session).
    assert download_service_mock.download_resource.call_count == 2
    # And the second item completed successfully — proves the session was
    # actually clean (rollback worked, not just was called).
    assert '"file": "B", "status": "success"' in body


# ---------------------------------------------------------------------------
# Change 3: the resource-level except in `_process_user_documents` calls
# db.rollback() at the TOP of the block. Drives the relevant path of the
# scheduler with focused mocks and verifies the rollback fires.
# ---------------------------------------------------------------------------


def test_process_user_documents_rolls_back_on_resource_error():
    """When `download_as_text` raises while processing a resource inside
    `_process_user_documents`, the per-resource except handler must call
    `db.rollback()` so the next resource's queries don't cascade on a
    poisoned session.
    """
    from datetime import datetime, UTC

    from local_deep_research.scheduler.background import (
        BackgroundJobScheduler,
        DocumentSchedulerSettings,
    )

    # Reset the singleton so we get a fresh scheduler.
    BackgroundJobScheduler._instance = None
    with patch("local_deep_research.scheduler.background.BackgroundScheduler"):
        scheduler = BackgroundJobScheduler()

    scheduler.user_sessions["testuser"] = {
        "scheduled_jobs": set(),
        "last_activity": datetime.now(UTC),
    }
    scheduler._credential_store.store("testuser", "testpass")

    settings = DocumentSchedulerSettings(
        download_pdfs=False,
        extract_text=True,
        generate_rag=False,
        last_run="",
    )

    # Build a minimal research session with two downloadable resources.
    research = MagicMock()
    research.id = "research-1"
    research.title = "T"
    research.completed_at = None
    resource_a = MagicMock()
    resource_a.id = 1
    resource_a.url = "https://example.com/a.pdf"
    resource_b = MagicMock()
    resource_b.id = 2
    resource_b.url = "https://example.com/b.pdf"

    db = MagicMock()
    research_query = MagicMock()
    research_query.filter.return_value = research_query
    research_query.order_by.return_value = research_query
    research_query.limit.return_value = research_query
    research_query.all.return_value = [research]
    resource_query = MagicMock()
    resource_query.filter_by.return_value = resource_query
    resource_query.all.return_value = [resource_a, resource_b]
    # First db.query() = ResearchHistory query, second = ResearchResource.
    db.query.side_effect = [research_query, resource_query]

    @MagicMock
    def _ctx():
        pass

    from contextlib import contextmanager as _cm

    @_cm
    def fake_get_user_db_session(*a, **kw):
        yield db

    download_service = MagicMock()
    download_service.__enter__ = MagicMock(return_value=download_service)
    download_service.__exit__ = MagicMock(return_value=False)
    # First resource raises, second succeeds. With the rollback in place,
    # the second call must complete normally.
    download_service.download_as_text.side_effect = [
        IntegrityError("INSERT", {}, Exception("UNIQUE constraint failed")),
        (True, None),
    ]

    settings_manager = MagicMock()

    with (
        patch.object(
            scheduler,
            "_get_document_scheduler_settings",
            return_value=settings,
        ),
        patch(
            "local_deep_research.database.session_context.get_user_db_session",
            side_effect=fake_get_user_db_session,
        ),
        patch(
            "local_deep_research.settings.manager.SettingsManager",
            return_value=settings_manager,
        ),
        patch(
            "local_deep_research.research_library.services.download_service.DownloadService",
            return_value=download_service,
        ),
        patch(
            "local_deep_research.research_library.utils.is_downloadable_url",
            return_value=True,
        ),
    ):
        scheduler._process_user_documents("testuser")

    # The per-resource except at line ~921 must have called db.rollback()
    # so the second resource's call could proceed on a clean session.
    assert db.rollback.called, (
        "db.rollback() must fire in the resource-level except so the loop "
        "can continue without PendingRollbackError on the next iteration."
    )
    # Both resources were attempted.
    assert download_service.download_as_text.call_count == 2


# ---------------------------------------------------------------------------
# Change 3 (continued): the text-extraction WRAPPER except at ~line 944 also
# calls safe_rollback. Triggered when an exception escapes the inner
# per-resource handling — e.g. the `db.query(ResearchResource).filter_by(...)`
# call itself raises before the per-resource loop starts.
# ---------------------------------------------------------------------------


def test_process_user_documents_rolls_back_on_text_extraction_wrapper_error():
    """If the text-extraction block raises before the per-resource loop —
    e.g. the resources query itself fails — the WRAPPER except must roll back,
    otherwise the post-loop last_run commit runs on a poisoned session. (RAG
    indexing has moved to _reconcile_unindexed_documents, but text-extraction
    rollback is still required for the last_run commit.)
    """
    from datetime import datetime, UTC

    from local_deep_research.scheduler.background import (
        BackgroundJobScheduler,
        DocumentSchedulerSettings,
    )

    BackgroundJobScheduler._instance = None
    with patch("local_deep_research.scheduler.background.BackgroundScheduler"):
        scheduler = BackgroundJobScheduler()

    scheduler.user_sessions["testuser"] = {
        "scheduled_jobs": set(),
        "last_activity": datetime.now(UTC),
    }
    scheduler._credential_store.store("testuser", "testpass")

    settings = DocumentSchedulerSettings(
        download_pdfs=False,
        extract_text=True,
        generate_rag=False,
        last_run="",
    )

    research = MagicMock()
    research.id = "research-x"
    research.title = "T"
    research.completed_at = None

    db = MagicMock()
    research_query = MagicMock()
    research_query.filter.return_value = research_query
    research_query.order_by.return_value = research_query
    research_query.limit.return_value = research_query
    research_query.all.return_value = [research]
    # Second db.query() call (for ResearchResource) — RAISES, which escapes
    # the inner per-resource try/except entirely and lands in the wrapper
    # at line ~944.
    resource_query_failure = RuntimeError("simulated DB query failure")
    db.query.side_effect = [research_query, resource_query_failure]

    from contextlib import contextmanager as _cm

    @_cm
    def fake_get_user_db_session(*a, **kw):
        yield db

    download_service = MagicMock()
    download_service.__enter__ = MagicMock(return_value=download_service)
    download_service.__exit__ = MagicMock(return_value=False)

    with (
        patch.object(
            scheduler,
            "_get_document_scheduler_settings",
            return_value=settings,
        ),
        patch(
            "local_deep_research.database.session_context.get_user_db_session",
            side_effect=fake_get_user_db_session,
        ),
        patch(
            "local_deep_research.settings.manager.SettingsManager",
            return_value=MagicMock(),
        ),
        patch(
            "local_deep_research.research_library.services.download_service.DownloadService",
            return_value=download_service,
        ),
    ):
        scheduler._process_user_documents("testuser")

    # The wrapper at line ~944 must have rolled back so any later code
    # (RAG, last_run commit) sees a clean session.
    assert db.rollback.called, (
        "db.rollback() must fire in the text-extraction wrapper except at "
        "line ~944 so subsequent phases run on a clean session."
    )


# ---------------------------------------------------------------------------
# Change 3 (continued): RAG indexing of research downloads has been RETIRED from
# _process_user_documents. It now lives in the unified
# _reconcile_unindexed_documents reconciler (its own scheduled job). These tests
# assert the inline RAG block is gone: generate_rag no longer drives any RAG
# work in this download/extract pass, and the module no longer imports
# LibraryRAGService at all.
# ---------------------------------------------------------------------------


def test_process_user_documents_no_longer_indexes_rag_inline():
    """With generate_rag=True but download/extract OFF, _process_user_documents
    short-circuits (no download/extract work) and never builds a RAG service —
    the inline RAG-indexing block has been retired into the reconciler. There is
    no longer a RAG wrapper rollback path here to exercise.
    """
    from datetime import datetime, UTC

    import local_deep_research.scheduler.background as bg
    from local_deep_research.scheduler.background import (
        BackgroundJobScheduler,
        DocumentSchedulerSettings,
    )

    # The retired block was the only consumer of LibraryRAGService at module
    # scope; its removal must hold so a regression can't quietly reintroduce
    # inline indexing.
    assert not hasattr(bg, "LibraryRAGService"), (
        "LibraryRAGService should no longer be imported in background.py — "
        "RAG indexing moved to _reconcile_unindexed_documents."
    )

    BackgroundJobScheduler._instance = None
    with patch("local_deep_research.scheduler.background.BackgroundScheduler"):
        scheduler = BackgroundJobScheduler()

    scheduler.user_sessions["testuser"] = {
        "scheduled_jobs": set(),
        "last_activity": datetime.now(UTC),
    }
    scheduler._credential_store.store("testuser", "testpass")

    settings = DocumentSchedulerSettings(
        download_pdfs=False,
        extract_text=False,
        generate_rag=True,
        last_run="",
    )

    from contextlib import contextmanager as _cm

    db = MagicMock()

    @_cm
    def fake_get_user_db_session(*a, **kw):
        yield db

    with (
        patch.object(
            scheduler,
            "_get_document_scheduler_settings",
            return_value=settings,
        ),
        patch(
            "local_deep_research.database.session_context.get_user_db_session",
            side_effect=fake_get_user_db_session,
        ) as mock_session,
        patch(
            "local_deep_research.settings.manager.SettingsManager",
            return_value=MagicMock(),
        ),
    ):
        scheduler._process_user_documents("testuser")

    # generate_rag alone no longer enables this pass: it short-circuits before
    # opening a DB session (download/extract are both off).
    mock_session.assert_not_called()
