diff --git a/learning_resources/hooks.py b/learning_resources/hooks.py index 36eb397094..e174bcf89b 100644 --- a/learning_resources/hooks.py +++ b/learning_resources/hooks.py @@ -82,7 +82,12 @@ def offeror_delete(self, offeror): @hookspec def content_files_loaded(self, run): - """Trigger actions after content files are loaded for a run""" + """ + Trigger actions after content files are loaded for a run. + + Args: + run: the LearningResourceRun whose content files were loaded + """ def get_plugin_manager(): diff --git a/learning_resources/utils.py b/learning_resources/utils.py index 01bc3947a7..3f0cd4752c 100644 --- a/learning_resources/utils.py +++ b/learning_resources/utils.py @@ -437,7 +437,10 @@ def bulk_resources_unpublished_actions(resource_ids: list[int], resource_type: s def content_files_loaded_actions(run: LearningResourceRun): """ - Trigger plugins when content files are loaded for a LearningResourceRun + Trigger plugins when content files are loaded for a LearningResourceRun. + + Args: + run: the LearningResourceRun whose content files were loaded """ pm = get_plugin_manager() hook = pm.hook diff --git a/learning_resources_search/plugins.py b/learning_resources_search/plugins.py index 7b62f42fcd..2a9ef9a223 100644 --- a/learning_resources_search/plugins.py +++ b/learning_resources_search/plugins.py @@ -261,12 +261,13 @@ def content_files_loaded(self, run): """ Upsert a created/modified run's content files. - Qdrant: embed every loaded run (all runs of a published/test_mode course) - and drop stale files. OpenSearch: index only the best published non-B2B - run, or any published non-variant run of a test_mode course. + Qdrant: embed the run's published files (unchanged files exit via the + checksum gate in vector_search) and drop stale files. OpenSearch: index + only the best published non-B2B run, or any published non-variant run + of a test_mode course. - Args: - run(LearningResourceRun): The LearningResourceRun that was upserted + Args: + run: the LearningResourceRun that was upserted """ if not run.content_files.exists(): return @@ -284,6 +285,8 @@ def content_files_loaded(self, run): if django_settings.QDRANT_ENABLE_INDEXING_PLUGIN_HOOKS: index_tasks.append(vector_tasks.embed_run_content_files.si(run.id)) + # Always purge unpublished files' points so a failed removal + # task self-heals on the next load. index_tasks.append( vector_tasks.remove_unpublished_run_content_files.si(run.id) ) diff --git a/learning_resources_search/plugins_test.py b/learning_resources_search/plugins_test.py index 6ee40c540f..e6b5f185b7 100644 --- a/learning_resources_search/plugins_test.py +++ b/learning_resources_search/plugins_test.py @@ -568,3 +568,21 @@ def test_search_index_plugin_resource_upserted_generate_embeddings( mock_search_index_helpers.mock_generate_embeddings_immutable_signature.assert_called_once_with( [resource.id], resource_type, overwrite=True ) + + +@pytest.mark.django_db +def test_content_files_loaded_always_purges_unpublished( + mock_search_index_helpers, settings +): + """The remove-unpublished task always runs so failed removals self-heal.""" + settings.QDRANT_ENABLE_INDEXING_PLUGIN_HOOKS = True + run = LearningResourceRunFactory.create( + published=True, learning_resource__create_runs=False + ) + ContentFileFactory.create(run=run) + + SearchIndexPlugin().content_files_loaded(run) + + mock_search_index_helpers.mock_remove_unpublished_run_contentfiles_immutable_signature.assert_called_once_with( + run.id + ) diff --git a/vector_search/constants.py b/vector_search/constants.py index 2302fae0bf..85ef1f42dc 100644 --- a/vector_search/constants.py +++ b/vector_search/constants.py @@ -5,6 +5,22 @@ CONTENT_FILES_COLLECTION_NAME = f"{settings.QDRANT_BASE_COLLECTION_NAME}.content_files" TOPICS_COLLECTION_NAME = f"{settings.QDRANT_BASE_COLLECTION_NAME}.topics" +# ContentFile columns (beyond checksum, which only covers content) compared by the +# embed_run_content_files pre-pass to detect stale Qdrant payloads. Every entry MUST +# be an exact serializer pass-through of a scalar/JSON ContentFile column: a field +# the serializer transforms would never converge, flagging every file on every load +# (test_content_file_prepass_fields_are_serializer_pass_through guards this). +CONTENT_FILE_PREPASS_PAYLOAD_FIELDS = ( + "title", + "description", + "url", + "file_type", + "file_extension", + "content_type", + "edx_module_id", + "summary", + "flashcards", +) QDRANT_CONTENT_FILE_PARAM_MAP = { "key": "key", diff --git a/vector_search/tasks.py b/vector_search/tasks.py index 4e842589bc..4559594033 100644 --- a/vector_search/tasks.py +++ b/vector_search/tasks.py @@ -15,6 +15,7 @@ ContentFile, Course, LearningResource, + LearningResourceRun, ) from learning_resources.serializers import ( ContentFileSerializer, @@ -39,10 +40,12 @@ now_in_utc, ) from vector_search.constants import ( + CONTENT_FILE_PREPASS_PAYLOAD_FIELDS, CONTENT_FILES_COLLECTION_NAME, RESOURCES_COLLECTION_NAME, ) from vector_search.utils import ( + _stored_content_payloads, embed_learning_resources, embed_topics, filter_existing_qdrant_points_by_ids, @@ -465,13 +468,76 @@ def embed_new_content_files(self): @app.task(bind=True) def embed_run_content_files(self, run_id): """ - Embed contentfiles associated with a run + Embed the run's published content files whose Qdrant points are missing or + stale (checksum or a payload metadata field differs). + + A run-level pre-pass batch-compares each file's DB checksum and payload + metadata columns against the stored Qdrant payload, so a fully-unchanged + run costs one DB query plus a few batched retrieves instead of serializing + every file. A checksum-matching file with drifted metadata (edited title, + newly generated summary, ...) is dispatched but exits via the payload-only + update path downstream — no re-embedding. Failed or purged embeds show up + as missing/stale points, so they self-heal on the next load. """ - content_file_ids = list( - ContentFile.objects.filter(run__id=run_id).values_list("id", flat=True) + run = ( + LearningResourceRun.objects.select_related("learning_resource__platform") + .filter(id=run_id) + .first() + ) + if run is None: + return None + resource = run.learning_resource + platform_code = resource.platform.code if resource.platform else "" + + def chunk0_point_id(key): + # Mirrors the doc fields ContentFileSerializer emits for run files + return vector_point_id( + vector_point_key( + { + "platform": {"code": platform_code}, + "resource_readable_id": resource.readable_id, + "run_readable_id": run.run_id, + "key": key, + }, + chunk_number=0, + document_type="content_file", + ) + ) + + pid_rows = [ + (cf_id, chunk0_point_id(key), checksum, meta) + for cf_id, key, checksum, *meta in ContentFile.objects.filter( + run=run, published=True + ).values_list("id", "key", "checksum", *CONTENT_FILE_PREPASS_PAYLOAD_FIELDS) + ] + stored = _stored_content_payloads( + [pid for _, pid, _, _ in pid_rows], + fields=("checksum", *CONTENT_FILE_PREPASS_PAYLOAD_FIELDS), ) - return _replace_with_finalized_chain(self, content_file_ids, overwrite=True) + def is_stale(pid, checksum, meta): + payload = stored.get(pid) + if payload is None or payload.get("checksum") != checksum: + return True + return any( + payload.get(field) != value + for field, value in zip(CONTENT_FILE_PREPASS_PAYLOAD_FIELDS, meta) + ) + + ids = [ + cf_id + for cf_id, pid, checksum, meta in pid_rows + if is_stale(pid, checksum, meta) + ] + log.info( + "embed_run_content_files run %s: %d of %d files need embedding", + run_id, + len(ids), + len(pid_rows), + ) + if not ids: + return None + return _replace_with_finalized_chain(self, ids, overwrite=True) @app.task(bind=True) diff --git a/vector_search/tasks_test.py b/vector_search/tasks_test.py index 769339ba19..632d8c4b40 100644 --- a/vector_search/tasks_test.py +++ b/vector_search/tasks_test.py @@ -28,7 +28,9 @@ PROGRAM_TYPE, ) from learning_resources_search.exceptions import RetryError +from learning_resources_search.serializers import serialize_bulk_content_files from main.utils import now_in_utc +from vector_search.constants import CONTENT_FILE_PREPASS_PAYLOAD_FIELDS from vector_search.tasks import ( _record_embedding_failure, _retry_countdown, @@ -44,7 +46,7 @@ remove_unpublished_run_content_files, start_embed_resources, ) -from vector_search.utils import vector_point_id +from vector_search.utils import vector_point_id, vector_point_key pytestmark = pytest.mark.django_db @@ -767,6 +769,157 @@ def test_embed_run_content_files_no_files_returns_none(mocker, mocked_celery): mocked_celery.chain.assert_not_called() +def test_embed_run_content_files_skips_unpublished(mocker, mocked_celery, settings): + """Unpublished files are never embedded.""" + settings.QDRANT_CHUNK_SIZE = 50 + run = LearningResourceRunFactory.create() + published = ContentFileFactory.create(run=run, published=True) + ContentFileFactory.create(run=run, published=False) + generate_embeddings_mock = mocker.patch( + "vector_search.tasks.generate_embeddings", autospec=True + ) + + with pytest.raises(mocked_celery.replace_exception_class): + embed_run_content_files.delay(run.id) + + assert _embedded_content_file_ids(generate_embeddings_mock) == {published.id} + + +def _serializer_chunk0_pids(content_files): + """Chunk-0 point ids as the embed pipeline (serializer path) computes them""" + return { + doc["id"]: vector_point_id( + vector_point_key(doc, chunk_number=0, document_type="content_file") + ) + for doc in serialize_bulk_content_files([cf.id for cf in content_files]) + } + + +def _stored_payload_entry(content_file, **overrides): + """Build a stored-payload map entry matching the file's current DB state""" + return { + "checksum": content_file.checksum, + **{ + field: getattr(content_file, field) + for field in CONTENT_FILE_PREPASS_PAYLOAD_FIELDS + }, + **overrides, + } + + +def test_embed_run_content_files_pre_pass_skips_unchanged( + mocker, mocked_celery, settings +): + """ + Only files whose stored Qdrant payload is missing or stale are embedded. + + The stored-payload map is keyed by serializer-derived point ids, so the + unchanged file is skipped only if the task's pre-pass computes the same + point id as the embed pipeline. + """ + settings.QDRANT_CHUNK_SIZE = 50 + run = LearningResourceRunFactory.create() + # ContentFile.save() computes checksum from content + unchanged = ContentFileFactory.create(run=run, published=True, content="aaa") + stale = ContentFileFactory.create(run=run, published=True, content="bbb") + missing = ContentFileFactory.create(run=run, published=True, content="ccc") + pids = _serializer_chunk0_pids([unchanged, stale, missing]) + stored_mock = mocker.patch( + "vector_search.tasks._stored_content_payloads", + return_value={ + pids[unchanged.id]: _stored_payload_entry(unchanged), + pids[stale.id]: _stored_payload_entry(stale, checksum="stale-checksum"), + }, + ) + generate_embeddings_mock = mocker.patch( + "vector_search.tasks.generate_embeddings", autospec=True + ) + + with pytest.raises(mocked_celery.replace_exception_class): + embed_run_content_files.delay(run.id) + + assert _embedded_content_file_ids(generate_embeddings_mock) == { + stale.id, + missing.id, + } + assert set(stored_mock.call_args.args[0]) == set(pids.values()) + + +def test_embed_run_content_files_pre_pass_dispatches_metadata_only_change( + mocker, mocked_celery, settings +): + """ + A file with a matching checksum but drifted payload metadata (edited title, + newly generated summary) is dispatched so its Qdrant payload gets refreshed. + """ + settings.QDRANT_CHUNK_SIZE = 50 + run = LearningResourceRunFactory.create() + retitled = ContentFileFactory.create(run=run, published=True, content="aaa") + summarized = ContentFileFactory.create(run=run, published=True, content="bbb") + current = ContentFileFactory.create(run=run, published=True, content="ccc") + pids = _serializer_chunk0_pids([retitled, summarized, current]) + mocker.patch( + "vector_search.tasks._stored_content_payloads", + return_value={ + pids[retitled.id]: _stored_payload_entry(retitled, title="old title"), + pids[summarized.id]: _stored_payload_entry(summarized, summary=""), + pids[current.id]: _stored_payload_entry(current), + }, + ) + summarized.summary = "a new summary" + summarized.save() + generate_embeddings_mock = mocker.patch( + "vector_search.tasks.generate_embeddings", autospec=True + ) + + with pytest.raises(mocked_celery.replace_exception_class): + embed_run_content_files.delay(run.id) + + assert _embedded_content_file_ids(generate_embeddings_mock) == { + retitled.id, + summarized.id, + } + + +def test_content_file_prepass_fields_are_serializer_pass_through(): + """ + Every pre-pass-compared field must be an exact serializer pass-through of + the ContentFile column: a transformed field would never converge with the + stored payload, flagging every file on every load. + """ + content_file = ContentFileFactory.create( + run=LearningResourceRunFactory.create(), + published=True, + content="some content", + summary="a summary", + flashcards=[{"question": "q", "answer": "a"}], + ) + doc = next(iter(serialize_bulk_content_files([content_file.id]))) + for field in ("checksum", *CONTENT_FILE_PREPASS_PAYLOAD_FIELDS): + assert doc[field] == getattr(content_file, field), field + + +def test_embed_run_content_files_all_unchanged_dispatches_nothing( + mocker, mocked_celery +): + """A fully-unchanged run embeds nothing and schedules no chain.""" + run = LearningResourceRunFactory.create() + files = ContentFileFactory.create_batch(2, run=run, published=True, content="x") + pids = _serializer_chunk0_pids(files) + mocker.patch( + "vector_search.tasks._stored_content_payloads", + return_value={pids[cf.id]: _stored_payload_entry(cf) for cf in files}, + ) + generate_embeddings_mock = mocker.patch( + "vector_search.tasks.generate_embeddings", autospec=True + ) + + assert embed_run_content_files(run.id) is None + + generate_embeddings_mock.si.assert_not_called() + mocked_celery.chain.assert_not_called() + + def test_embeddings_healthcheck_no_missing_embeddings(mocker): """ Test embeddings_healthcheck when there are no missing embeddings diff --git a/vector_search/utils.py b/vector_search/utils.py index 67aab4ee5f..0665a2dfc9 100644 --- a/vector_search/utils.py +++ b/vector_search/utils.py @@ -637,17 +637,29 @@ def _retrieve_content_file_point( return None -def _content_file_stored_checksum_changed(serialized_document: dict) -> bool: - point = _retrieve_content_file_point(serialized_document) - if not point: - return False - stored_checksum = (point.payload or {}).get("checksum") - # Missing checksums should not force an expensive summary rewrite by themselves. - # should_generate_content_embeddings still treats them as changed so embeddings - # can repair older Qdrant points without overwriting existing summaries. - return stored_checksum is not None and stored_checksum != serialized_document.get( - "checksum" - ) +def _stored_content_payloads( + point_ids: list[str], fields: tuple[str, ...] = ("checksum",) +) -> dict[str, dict]: + """ + Batch-retrieve stored payload fields for content-file points. + + Returns {point_id: partial payload dict} for points that exist in Qdrant; + absent points are absent from the map. One lookup per batch replaces the + per-file retrieves for the existence filter, summary-change check, and + embed gate. + """ + client = qdrant_client() + stored = {} + for id_batch in chunks( + point_ids, chunk_size=settings.QDRANT_POINT_UPLOAD_BATCH_SIZE + ): + for record in client.retrieve( + collection_name=CONTENT_FILES_COLLECTION_NAME, + ids=id_batch, + with_payload=list(fields), + ): + stored[record.id] = record.payload or {} + return stored def should_generate_content_embeddings( @@ -743,9 +755,13 @@ def _embed_course_metadata_as_contentfile(serialized_resources): client.upload_points(CONTENT_FILES_COLLECTION_NAME, points=points, wait=False) -def _generate_content_file_points(serialized_content): +def _generate_content_file_points(serialized_content, stored_payloads): """ - Chunk and embed content file documents, yielding PointStructs + Chunk and embed content file documents, yielding PointStructs. + + stored_payloads maps chunk-0 point ids to stored Qdrant payload fields + (see _stored_content_payloads); docs whose stored checksum matches get a + payload-only refresh instead of re-embedding. """ encoder_dense = dense_encoder() encoder_sparse = sparse_encoder() @@ -771,7 +787,17 @@ def _generate_content_file_points(serialized_content): embedding_context = _content_file_embedding_context(doc) if not embedding_context: continue - should_generate = should_generate_content_embeddings(doc) + # Point ids are content-key-derived and stable, so recompute per doc; + # summarization replaces the doc dicts between here and process_batch. + point_id = vector_point_id( + vector_point_key(doc, chunk_number=0, document_type="content_file") + ) + # Missing point or differing/missing stored checksum -> regenerate + # (self-heals failed or purged points on the next load). + should_generate = ( + point_id not in stored_payloads + or stored_payloads[point_id].get("checksum") != doc["checksum"] + ) if not should_generate: """ Just update the payload and continue @@ -947,7 +973,6 @@ def process_batch(docs_batch): fill_summary_content_ids = [] changed_summary_content_ids = [] - # Collect IDs for summarization contentfile_points = [ ( vector_point_id( @@ -959,17 +984,23 @@ def process_batch(docs_batch): ) for doc in docs_batch ] + # One batched lookup serves the existence filter, the summary-change + # check, and the embed gate in _generate_content_file_points. + stored_payloads = _stored_content_payloads( + [point[0] for point in contentfile_points] + ) if not overwrite: - filtered_point_ids = filter_existing_qdrant_points_by_ids( - [point[0] for point in contentfile_points], - collection_name=collection_name, - ) docs_batch = [ - point[1] + doc + for point_id, doc in contentfile_points + if point_id not in stored_payloads + ] + contentfile_points = [ + point for point in contentfile_points - if point[0] in filtered_point_ids + if point[0] not in stored_payloads ] - for resource in docs_batch: + for point_id, resource in contentfile_points: if ( resource.get("summary") or resource.get("require_summaries") @@ -977,7 +1008,15 @@ def process_batch(docs_batch): .filter(run__id=resource.get("run_id")) .exists() ): - if overwrite and _content_file_stored_checksum_changed(resource): + stored_checksum = stored_payloads.get(point_id, {}).get("checksum") + # A missing point or missing stored checksum must not force + # an expensive summary rewrite by itself; the embed gate + # still regenerates embeddings for those. + if ( + overwrite + and stored_checksum is not None + and stored_checksum != resource.get("checksum") + ): changed_summary_content_ids.append(resource["id"]) else: fill_summary_content_ids.append(resource["id"]) @@ -988,7 +1027,9 @@ def process_batch(docs_batch): changed_summary_content_ids, ) - points_generator_iter = _generate_content_file_points(docs_batch) + points_generator_iter = _generate_content_file_points( + docs_batch, stored_payloads + ) points_upload_batch = [] for point in points_generator_iter: diff --git a/vector_search/utils_test.py b/vector_search/utils_test.py index 28ed8b5226..af5a9cb346 100644 --- a/vector_search/utils_test.py +++ b/vector_search/utils_test.py @@ -226,16 +226,16 @@ def test_embed_learning_resources_no_overwrite(mocker, content_type): ], ) else: - # all contentfiles exist in qdrant - mocker.patch( - "vector_search.utils.filter_existing_qdrant_points_by_ids", - return_value=[ - vector_point_id( - f"{doc['platform']['code']}.{doc['resource_readable_id']}.{doc['run_readable_id']}.{doc['key']}.0" - ) - for doc in serialize_bulk_content_files([r.id for r in resources[0:3]]) - ], - ) + # the last 2 contentfiles already have points in qdrant; the first 3 don't + mock_qdrant.retrieve.return_value = [ + mocker.MagicMock( + id=vector_point_id( + vector_point_key(doc, chunk_number=0, document_type="content_file") + ), + payload={"checksum": doc["checksum"]}, + ) + for doc in serialize_bulk_content_files([r.id for r in resources[3:5]]) + ] mocker.patch( "learning_resources.content_summarizer.ContentSummarizer.summarize_content_files_by_ids" ) @@ -770,9 +770,6 @@ def test_generate_content_points_uses_markdown_chunking_for_marketing_pages(mock return_value=[Document(page_content="chunk1", metadata={"key": "k1"})], ) mock_chunk = mocker.patch("vector_search.utils._chunk_documents") - mocker.patch( - "vector_search.utils.should_generate_content_embeddings", return_value=True - ) mocker.patch("vector_search.utils.remove_points_matching_params") mock_dense = mocker.MagicMock() @@ -794,7 +791,7 @@ def test_generate_content_points_uses_markdown_chunking_for_marketing_pages(mock "key": "k1", } - list(_generate_content_file_points([doc])) + list(_generate_content_file_points([doc], {})) mock_md_chunk.assert_called_once() mock_chunk.assert_not_called() @@ -810,9 +807,6 @@ def test_generate_content_points_uses_standard_chunking_for_non_markdown(mocker) "vector_search.utils._chunk_documents", return_value=[Document(page_content="chunk1", metadata={"key": "k1"})], ) - mocker.patch( - "vector_search.utils.should_generate_content_embeddings", return_value=True - ) mocker.patch("vector_search.utils.remove_points_matching_params") mock_dense = mocker.MagicMock() @@ -834,7 +828,7 @@ def test_generate_content_points_uses_standard_chunking_for_non_markdown(mocker) "key": "k1", } - list(_generate_content_file_points([doc])) + list(_generate_content_file_points([doc], {})) mock_chunk.assert_called_once() mock_md_chunk.assert_not_called() @@ -857,9 +851,6 @@ def test_generate_content_points_leaves_headroom_under_token_limit(mocker): for i in range(num_chunks) ], ) - mocker.patch( - "vector_search.utils.should_generate_content_embeddings", return_value=True - ) mocker.patch("vector_search.utils.remove_points_matching_params") mock_dense = mocker.MagicMock() @@ -881,7 +872,7 @@ def test_generate_content_points_leaves_headroom_under_token_limit(mocker): "key": "k1", } - points = list(_generate_content_file_points([doc])) + points = list(_generate_content_file_points([doc], {})) batch_sizes = [ len(call.args[0]) for call in mock_dense.embed_documents.call_args_list @@ -906,9 +897,6 @@ def test_generate_content_points_request_chunk_size_never_zero(mocker): Document(page_content=f"chunk{i}", metadata={"key": "k1"}) for i in range(3) ], ) - mocker.patch( - "vector_search.utils.should_generate_content_embeddings", return_value=True - ) mocker.patch("vector_search.utils.remove_points_matching_params") mock_dense = mocker.MagicMock() @@ -930,7 +918,7 @@ def test_generate_content_points_request_chunk_size_never_zero(mocker): "key": "k1", } - points = list(_generate_content_file_points([doc])) + points = list(_generate_content_file_points([doc], {})) assert len(points) == 3 @@ -1127,43 +1115,81 @@ def test_should_generate_for_changed_content_file(mocker): assert result is True +def test_stored_content_payloads_batches_and_maps(mocker, settings): + """One retrieve per id-chunk; existing points map to their stored payload.""" + settings.QDRANT_POINT_UPLOAD_BATCH_SIZE = 2 + present = mocker.MagicMock(id="p1", payload={"checksum": "abc"}) + no_checksum = mocker.MagicMock(id="p2", payload={}) + mock_qdrant = mocker.MagicMock() + # p3 does not exist in Qdrant + mock_qdrant.retrieve.side_effect = [[present, no_checksum], []] + mocker.patch("vector_search.utils.qdrant_client", return_value=mock_qdrant) + + stored = vs_utils._stored_content_payloads( # noqa: SLF001 + ["p1", "p2", "p3"], fields=("checksum", "title") + ) + + assert stored == {"p1": {"checksum": "abc"}, "p2": {}} + assert mock_qdrant.retrieve.call_count == 2 # ceil(3 ids / batch size 2) + for call in mock_qdrant.retrieve.call_args_list: + assert call.kwargs["collection_name"] == CONTENT_FILES_COLLECTION_NAME + assert call.kwargs["with_payload"] == ["checksum", "title"] + + @pytest.mark.parametrize( - ("stored_payload", "expected"), + ("stored_entry", "expect_regenerate"), [ - ({"checksum": "previous-checksum"}, True), - ({"checksum": "current-checksum"}, False), - ({}, False), - (None, False), + ("missing", True), # no point in Qdrant (new file or failed prior embed) + (None, True), # point exists but has no stored checksum + ("stale-checksum", True), # stored checksum differs + ("current-checksum", False), # matches -> payload-only update ], -) -def test_content_file_stored_checksum_changed(mocker, stored_payload, expected): - """Only an existing, different stored checksum counts as changed for summaries.""" - serialized_document = { - "resource_readable_id": "resource-1", - "run_readable_id": "run-1", - "key": "transcript.txt", - "checksum": "current-checksum", - } - mock_qdrant = mocker.MagicMock() - if stored_payload is None: - mock_qdrant.retrieve.return_value = [] - else: - mock_point = mocker.MagicMock() - mock_point.payload = stored_payload - mock_qdrant.retrieve.return_value = [mock_point] - mocker.patch("vector_search.utils.qdrant_client", return_value=mock_qdrant) +) # stored_entry is the checksum in the stored payload dict +def test_generate_content_points_checksum_gate(mocker, stored_entry, expect_regenerate): + """Docs are re-embedded unless their stored Qdrant checksum matches.""" + settings.CONTENT_FILE_EMBEDDING_CHUNK_SIZE_OVERRIDE = 500 + settings.CONTENT_FILE_EMBEDDING_CHUNK_OVERLAP = 50 - assert ( - vs_utils._content_file_stored_checksum_changed( # noqa: SLF001 - serialized_document - ) - is expected + mocker.patch( + "vector_search.utils._chunk_documents", + return_value=[Document(page_content="chunk1", metadata={"key": "k1"})], ) - mock_qdrant.retrieve.assert_called_once() - assert ( - mock_qdrant.retrieve.call_args.kwargs["collection_name"] - == CONTENT_FILES_COLLECTION_NAME + mocker.patch("vector_search.utils.remove_points_matching_params") + update_payload_mock = mocker.patch( + "vector_search.utils.update_content_file_payload" + ) + mock_dense = mocker.MagicMock() + mock_dense.embed_documents.side_effect = lambda texts: [[0.1] for _ in texts] + mock_dense.model_short_name.return_value = "dense" + mock_sparse = mocker.MagicMock() + mock_sparse.embed_documents.side_effect = lambda texts: [[0.2] for _ in texts] + mock_sparse.model_short_name.return_value = "sparse" + mocker.patch("vector_search.utils.dense_encoder", return_value=mock_dense) + mocker.patch("vector_search.utils.sparse_encoder", return_value=mock_sparse) + + doc = { + "content": "Some plain text content", + "file_type": "page", + "file_extension": ".html", + "platform": {"code": "x"}, + "resource_readable_id": "r1", + "run_readable_id": "run1", + "key": "k1", + "checksum": "current-checksum", + } + point_id = vector_point_id( + vector_point_key(doc, chunk_number=0, document_type="content_file") ) + stored = {} if stored_entry == "missing" else {point_id: {"checksum": stored_entry}} + + points = list(_generate_content_file_points([doc], stored)) + + if expect_regenerate: + assert len(points) == 1 + update_payload_mock.assert_not_called() + else: + assert points == [] + update_payload_mock.assert_called_once_with(doc) def test_should_not_generate_for_unchanged_content_file(mocker): @@ -1246,11 +1272,9 @@ def test_embed_learning_resources_summarizes_only_contentfiles_with_summary(mock Test that embedding overwrites don't overwrite existing summaries. """ mock_qdrant = mocker.patch("qdrant_client.QdrantClient") + mock_qdrant.retrieve.return_value = [] mocker.patch("vector_search.utils.qdrant_client", return_value=mock_qdrant) mocker.patch("vector_search.utils.create_qdrant_collections") - mocker.patch( - "vector_search.utils.filter_existing_qdrant_points_by_ids", return_value=[] - ) mocker.patch("vector_search.utils.remove_qdrant_records") learning_resource = LearningResourceFactory.create( @@ -1285,9 +1309,6 @@ def test_embed_learning_resources_summarizes_only_contentfiles_with_summary(mock mocker.patch( "vector_search.utils.serialize_bulk_content_files", return_value=serialized ) - mocker.patch( - "vector_search.utils._content_file_stored_checksum_changed", return_value=False - ) summarize_mock = mocker.patch( "learning_resources.content_summarizer.ContentSummarizer.summarize_content_files_by_ids" @@ -1333,7 +1354,7 @@ def test_embed_learning_resources_overwrites_summaries_for_changed_content(mocke "key": cf.key, "summary": cf.summary, "content": cf.content, - "checksum": cf.checksum, + "checksum": f"current-{cf.id}", } for cf in all_contentfiles ] @@ -1341,10 +1362,22 @@ def test_embed_learning_resources_overwrites_summaries_for_changed_content(mocke mocker.patch( "vector_search.utils.serialize_bulk_content_files", return_value=serialized ) - mocker.patch( - "vector_search.utils._content_file_stored_checksum_changed", - side_effect=lambda resource: resource["id"] == changed_content_file.id, - ) + # The unchanged file takes the payload-only path (covered by its own tests) + mocker.patch("vector_search.utils.update_content_file_payload") + # Stored Qdrant checksum matches for the unchanged file, differs for the changed + mock_qdrant.retrieve.return_value = [ + mocker.MagicMock( + id=vector_point_id( + vector_point_key(doc, chunk_number=0, document_type="content_file") + ), + payload={ + "checksum": doc["checksum"] + if doc["id"] == unchanged_content_file.id + else "stale-checksum" + }, + ) + for doc in serialized + ] summarize_mock = mocker.patch( "learning_resources.content_summarizer.ContentSummarizer.summarize_content_files_by_ids" @@ -1392,15 +1425,23 @@ def test_embed_learning_resources_keeps_old_checksum_when_summary_fails(mocker): "key": content_file.key, "summary": content_file.summary, "content": content_file.content, - "checksum": content_file.checksum, + "checksum": "current-checksum", } ] mocker.patch( "vector_search.utils.serialize_bulk_content_files", return_value=serialized ) - mocker.patch( - "vector_search.utils._content_file_stored_checksum_changed", return_value=True - ) + # Stored Qdrant checksum differs, so the summary must be regenerated + mock_qdrant.retrieve.return_value = [ + mocker.MagicMock( + id=vector_point_id( + vector_point_key( + serialized[0], chunk_number=0, document_type="content_file" + ) + ), + payload={"checksum": "previous-checksum"}, + ) + ] summarize_mock = mocker.patch( "learning_resources.content_summarizer.ContentSummarizer.summarize_content_files_by_ids", return_value=[