diff --git a/bertopic/_bertopic.py b/bertopic/_bertopic.py
index cfafb58a..05f0f10c 100644
--- a/bertopic/_bertopic.py
+++ b/bertopic/_bertopic.py
@@ -1,5 +1,6 @@
 # ruff: noqa: E402
 import yaml
+import hashlib
 import warnings
 
 warnings.filterwarnings("ignore", category=FutureWarning)
@@ -4232,6 +4233,31 @@ def _save_representative_docs(self, documents: pd.DataFrame):
         )
         self.representative_docs_ = repr_docs
 
+    @staticmethod
+    def _image_dedup_key(image) -> str | int:
+        """Hashable stand-in for an image, for use as a `drop_duplicates` subset key.
+
+        `drop_duplicates` hashes its subset columns, but PIL sets `Image.__hash__ =
+        None`, so a loaded `Image` can't be used as a key directly. Paths key on
+        themselves; loaded images key on their content (mode, size, pixel bytes),
+        matching PIL's own content-based `Image.__eq__` - so two pixel-identical
+        images collapse just as PIL considers them equal, while distinct images
+        (even with an identical caption) do not.
+
+        Arguments:
+            image: An image path (`str`) or a loaded image object (e.g. `PIL.Image`).
+
+        Returns:
+            A hashable key such that two images compare equal under this key iff
+            they should be treated as duplicates.
+        """
+        if isinstance(image, str):
+            return image
+        to_bytes = getattr(image, "tobytes", None)  # duck-typed: Pillow is an optional dep
+        if to_bytes is None:
+            return id(image)
+        return f"{image.mode}|{image.size}|{hashlib.sha1(to_bytes()).hexdigest()}"
+
     def _extract_representative_docs(
         self,
         c_tf_idf: csr_matrix,
@@ -4240,7 +4266,7 @@ def _extract_representative_docs(
         nr_samples: int = 500,
         nr_repr_docs: int = 5,
         diversity: float | None = None,
-    ) -> Union[List[str], List[List[int]]]:
+    ) -> Tuple[Mapping[int, List[str]], List[str], List[List[int]], List[List[int]]]:
         """Approximate most representative documents per topic by sampling
         a subset of the documents in each topic and calculating which are
         most representative to their topic based on the cosine similarity between
@@ -4258,18 +4284,52 @@ def _extract_representative_docs(
         Returns:
             repr_docs_mappings: A dictionary from topic to representative documents
             representative_docs: A flat list of representative documents
-            repr_doc_indices: Ordered indices of representative documents
-                              that belong to each topic
+            repr_doc_indices: Positions into the flat `repr_docs` list, grouped by topic
             repr_doc_ids: The indices of representative documents
                           that belong to each topic
         """
         # Sample documents per topic
+        # Dedup on (Topic, Document) first; `Image` isn't hashable (PIL sets
+        # `Image.__hash__ = None`), so it can't be a `drop_duplicates` subset column
+        # directly, and content-hashing every image up front would cost a full
+        # pixel-buffer read per document. Only rows that collide on (Topic, Document)
+        # need their images compared - a row with unique text is already unique - so
+        # `_image_dedup_key` is computed for those rows only. In the multimodal path,
+        # distinct images can produce identical captions in `Document`; without this,
+        # deduplicating on `Document` alone would silently collapse them into a single
+        # candidate, starving `VisualRepresentation` of images below `nr_repr_images`.
+        dedup_keys = pd.DataFrame({"Topic": documents["Topic"], "Document": documents["Document"]})
+        if "Image" in documents.columns:
+            duplicated = dedup_keys.duplicated(keep=False).to_numpy()
+            dedup_keys["Image"] = [
+                self._image_dedup_key(image) if is_duplicate else None
+                for is_duplicate, image in zip(duplicated, documents["Image"].to_numpy())
+            ]
+        deduplicated_documents = documents[~dedup_keys.duplicated()].drop("Image", axis=1, errors="ignore")
+
+        # Sample without replacement, capped at each topic's size. Shuffling the whole frame
+        # once and then taking each topic's first `nr_samples` rows draws exactly that: a
+        # uniform sample of `min(nr_samples, len(group))` rows per topic. `GroupBy.sample`
+        # cannot express the per-group cap - it raises when a group holds fewer rows than `n`,
+        # and `n` is a scalar in every pandas release - while a per-group Python loop costs a
+        # `sample` call and a `concat` block per topic, which is ~10x slower at a few thousand
+        # topics. `head` preserves the original document index, which `selection.index` below
+        # relies on. A single global shuffle also decorrelates topics by construction: a fixed
+        # seed applied per group would draw the same positional pattern for equally-sized
+        # topics, so their samples would agree on e.g. "first document, third document, ...".
         documents_per_topic = (
-            documents.drop("Image", axis=1, errors="ignore")
-            .groupby("Topic")
-            .sample(n=nr_samples, replace=True, random_state=42)
-            .drop_duplicates()
+            deduplicated_documents.sample(frac=1, random_state=42).groupby("Topic", sort=False).head(nr_samples)
         )
+        if documents_per_topic.empty:
+            # `groupby` silently drops NaN keys, so an empty `documents` or an all-NaN
+            # `Topic` column both leave nothing here. Without this guard the failure
+            # surfaces much later as an empty or partial result with no indication that
+            # the real cause is upstream: no document has a valid topic assignment yet.
+            raise ValueError(
+                "No documents with a valid `Topic` assignment were found to extract "
+                "representative documents from. This happens when `documents` is empty "
+                "or every document's `Topic` is NaN (topics have not been assigned yet)."
+            )
 
         # Find and extract documents that are most similar to the topic
         repr_docs = []
@@ -4291,24 +4351,32 @@ def _extract_representative_docs(
 
             # Use MMR to find representative but diverse documents
             if diversity:
-                docs = mmr(
+                # `mmr()` only inspects `word_embeddings`/`doc_embedding` for its
+                # similarity math; the `words` argument is returned as-is at the
+                # end (`[words[idx] for idx in keywords_idx]`) and is never used
+                # to compute anything. Passing positions instead of the document
+                # strings lets `mmr()` hand back the selected positions directly,
+                # so there is no text-keyed reverse lookup and therefore no
+                # assumption that `selected_docs` contains unique text.
+                selected_indices = mmr(
                     c_tf_idf[index],
                     ctfidf,
-                    selected_docs,
+                    list(range(len(selected_docs))),
                     top_n=nr_docs,
                     diversity=diversity,
                 )
+                docs = [selected_docs[i] for i in selected_indices]
 
             # Extract top n most representative documents
             else:
-                indices = np.argpartition(sim_matrix.reshape(1, -1)[0], -nr_docs)[-nr_docs:]
-                docs = [selected_docs[index] for index in indices]
+                selected_indices = np.argpartition(sim_matrix.reshape(1, -1)[0], -nr_docs)[-nr_docs:]
+                docs = [selected_docs[i] for i in selected_indices]
 
-            doc_ids = [selected_docs_ids[index] for index, doc in enumerate(selected_docs) if doc in docs]
+            doc_ids = [selected_docs_ids[i] for i in selected_indices]
             repr_docs_ids.append(doc_ids)
             repr_docs.extend(docs)
             repr_docs_indices.append([repr_docs_indices[-1][-1] + i + 1 if index != 0 else i for i in range(nr_docs)])
-        repr_docs_mappings = {topic: repr_docs[i[0] : i[-1] + 1] for topic, i in zip(topics.keys(), repr_docs_indices)}
+        repr_docs_mappings = {topic: repr_docs[i[0] : i[-1] + 1] for topic, i in zip(labels, repr_docs_indices)}
 
         return repr_docs_mappings, repr_docs, repr_docs_indices, repr_docs_ids
 
diff --git a/bertopic/representation/_visual.py b/bertopic/representation/_visual.py
index 8c98d5a6..b2d8bc42 100644
--- a/bertopic/representation/_visual.py
+++ b/bertopic/representation/_visual.py
@@ -92,7 +92,9 @@ def extract_topics(
             representative_images: Representative images per topic
         """
         # Extract image ids of most representative documents
-        images = documents["Image"].to_numpy().tolist()
+        # NOTE: `repr_docs_ids` (below) contains index labels, not positions, so keep
+        # `images` as a label-indexed Series rather than flattening it to a plain list.
+        images = documents["Image"]
         (_, _, _, repr_docs_ids) = topic_model._extract_representative_docs(
             c_tf_idf,
             documents,
@@ -110,8 +112,8 @@ def extract_topics(
             sliced_examplars = [sliced_examplars[i : i + 3] for i in range(0, len(sliced_examplars), 3)]
             images_to_combine = [
                 [
-                    Image.open(images[index]) if isinstance(images[index], str) else images[index]
-                    for index in sub_indices
+                    Image.open(img) if isinstance(img, str) else img
+                    for img in (images.loc[index] for index in sub_indices)
                 ]
                 for sub_indices in sliced_examplars
             ]
@@ -121,7 +123,7 @@ def extract_topics(
             representative_images[topic] = representative_image
 
             # Make sure to properly close images
-            if isinstance(images[0], str):
+            if isinstance(images.iloc[0], str):
                 for image_list in images_to_combine:
                     for image in image_list:
                         image.close()
diff --git a/tests/conftest.py b/tests/conftest.py
index fd278b0f..a461de15 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -1,4 +1,5 @@
 import copy
+import pandas as pd
 import pytest
 from umap import UMAP
 from hdbscan import HDBSCAN
@@ -7,12 +8,77 @@
 from sentence_transformers import SentenceTransformer
 from sklearn.cluster import KMeans, MiniBatchKMeans
 from sklearn.decomposition import PCA
-from bertopic.vectorizers import OnlineCountVectorizer
+from sklearn.feature_extraction.text import CountVectorizer
+from bertopic.vectorizers import OnlineCountVectorizer, ClassTfidfTransformer
 from bertopic.representation import KeyBERTInspired, MaximalMarginalRelevance
 from bertopic.dimensionality import BaseDimensionalityReduction
 from sklearn.linear_model import LogisticRegression
 
 
+@pytest.fixture
+def minimal_topic_model():
+    """Factory fixture building a network-free BERTopic model (vectorizer + c-TF-IDF only),
+    for exercising `_extract_representative_docs` directly without fitting embeddings/UMAP/HDBSCAN.
+
+    Args passed to the returned builder:
+        docs: list of document strings
+        topics_list: list of topic ids, one per doc, aligned with `docs`
+        index: optional custom index for the resulting `documents` DataFrame (defaults to a
+               default RangeIndex). Use a non-contiguous/shifted index to exercise label-based
+               (as opposed to positional) indexing.
+        ids: optional values for the `ID` column (defaults to `range(len(docs))`). Pass values
+             distinct from `index` to mirror the zero-shot path where `ID` is reset independently
+             of the DataFrame index.
+        topic_order: optional explicit key insertion order for the returned `topics` dict
+                      (defaults to sorted topic ids). Use a non-sorted order to exercise code
+                      that (incorrectly) relies on dict insertion order instead of sorted labels.
+        images: optional list of per-document image identifiers, aligned with `docs`, adding an
+                `Image` column. Use distinct values on rows that otherwise share `Document` text
+                to exercise the multimodal dedup path.
+
+    Returns: (model, c_tf_idf, documents, topics)
+    """
+
+    def _build(docs, topics_list, index=None, ids=None, topic_order=None, images=None):
+        documents = pd.DataFrame(
+            {
+                "Document": docs,
+                "ID": ids if ids is not None else range(len(docs)),
+                "Topic": topics_list,
+            }
+        )
+        if images is not None:
+            documents["Image"] = images
+        if index is not None:
+            documents.index = index
+
+        vectorizer = CountVectorizer()
+        docs_per_topic = documents.groupby(["Topic"], as_index=False).agg({"Document": " ".join})
+        X = vectorizer.fit_transform(docs_per_topic.Document.values)
+        ctfidf_model = ClassTfidfTransformer()
+        ctfidf_model.fit(X)
+        c_tf_idf = ctfidf_model.transform(X)
+
+        model = BERTopic()
+        model.vectorizer_model = vectorizer
+        model.ctfidf_model = ctfidf_model
+
+        order = topic_order if topic_order is not None else sorted(documents.Topic.unique())
+        topics = {}
+        for topic_id in order:
+            topic_docs = docs_per_topic.loc[docs_per_topic.Topic == topic_id, "Document"].to_numpy()[0]
+            bow = vectorizer.transform([topic_docs])
+            tf = ctfidf_model.transform(bow)
+            feature_names = vectorizer.get_feature_names_out()
+            scores = tf.toarray().flatten()
+            top_indices = scores.argsort()[-5:][::-1]
+            topics[topic_id] = [(feature_names[i], float(scores[i])) for i in top_indices]
+
+        return model, c_tf_idf, documents, topics
+
+    return _build
+
+
 @pytest.fixture(scope="session")
 def embedding_model():
     model = SentenceTransformer("all-MiniLM-L6-v2")
diff --git a/tests/test_dedup_representative_docs.py b/tests/test_dedup_representative_docs.py
new file mode 100644
index 00000000..953a0ba9
--- /dev/null
+++ b/tests/test_dedup_representative_docs.py
@@ -0,0 +1,246 @@
+"""Tests for deduplicating representative documents sampling.
+
+Verifies that `_extract_representative_docs` samples without replacement so a
+topic never yields duplicate representative documents.
+
+Run from BERTopic repo root:
+    pytest tests/test_dedup_representative_docs.py -v
+"""
+
+import pytest
+
+
+def test_no_duplicate_docs_per_topic(minimal_topic_model):
+    """Each topic's representative docs should contain no duplicates."""
+    docs = ["alpha", "beta", "gamma", "delta", "epsilon"] * 3
+    topics_list = [0, 0, 0, 1, 1] * 3
+
+    model, c_tf_idf, documents, topics = minimal_topic_model(docs, topics_list)
+
+    repr_docs_mappings, _repr_docs, _, _ = model._extract_representative_docs(
+        c_tf_idf,
+        documents,
+        topics,
+        nr_samples=500,
+        nr_repr_docs=5,
+    )
+
+    assert repr_docs_mappings
+    for topic, topic_docs in repr_docs_mappings.items():
+        assert len(topic_docs) == len(set(topic_docs)), (
+            f"Topic {topic} has duplicate representative docs: {[d for d in topic_docs if topic_docs.count(d) > 1]}"
+        )
+
+
+def test_heavy_duplicates_no_duplicates_in_output(minimal_topic_model):
+    """When a topic has 3 unique docs but nr_samples=500, no duplicates should appear."""
+    docs = ["doc A", "doc B", "doc C"] * 2 + ["unique doc"]
+    topics_list = [0, 0, 0, 1, 1, 1, 0]
+
+    model, c_tf_idf, documents, topics = minimal_topic_model(docs, topics_list)
+
+    repr_docs_mappings, _repr_docs, _, _ = model._extract_representative_docs(
+        c_tf_idf,
+        documents,
+        topics,
+        nr_samples=500,
+        nr_repr_docs=3,
+    )
+
+    for topic, docs_list in repr_docs_mappings.items():
+        assert len(docs_list) == len(set(docs_list)), f"Topic {topic} has duplicate representative docs"
+
+
+def test_repr_docs_count_respects_topic_size(minimal_topic_model):
+    """nr_repr_docs should be capped at the number of unique docs in the topic."""
+    docs = ["alpha", "beta", "gamma", "delta", "epsilon", "zeta"]
+    topics_list = [0, 0, 1, 1, 1, 1]
+    # Topic 0 has only 2 unique docs — requesting 5 should yield 2
+
+    model, c_tf_idf, documents, topics = minimal_topic_model(docs, topics_list)
+
+    repr_docs_mappings, _, _, _ = model._extract_representative_docs(
+        c_tf_idf,
+        documents,
+        topics,
+        nr_samples=500,
+        nr_repr_docs=5,
+    )
+
+    assert len(repr_docs_mappings[0]) == 2
+    assert len(repr_docs_mappings[1]) == 4
+
+
+def test_repr_docs_count_with_nr_repr_docs_greater_than_topic_size(minimal_topic_model):
+    """When nr_repr_docs > unique docs in a topic, return all unique docs."""
+    docs = ["only one"] * 5 + ["other topic doc"] * 5
+    topics_list = [0] * 5 + [1] * 5
+    # Topic 0 has 1 unique doc, topic 1 has 1 unique doc
+
+    model, c_tf_idf, documents, topics = minimal_topic_model(docs, topics_list)
+
+    repr_docs_mappings, _, _, _ = model._extract_representative_docs(
+        c_tf_idf,
+        documents,
+        topics,
+        nr_samples=500,
+        nr_repr_docs=10,
+    )
+
+    for topic, docs_list in repr_docs_mappings.items():
+        assert len(docs_list) == 1
+
+
+def test_nr_samples_caps_candidates_per_topic(minimal_topic_model):
+    """`nr_samples` must cap the candidate pool per topic, independently of topic size.
+
+    The cap is what makes this an *approximate* search: only `nr_samples` documents
+    per topic are scored. It is enforced by taking each topic's first `nr_samples`
+    rows from a globally shuffled frame, so - unlike the explicit `min(nr_samples,
+    len(group))` it replaced - nothing in the expression names the cap. With
+    `nr_samples=2` only 2 documents per topic are scored, so at most 2 can come back
+    even though `nr_repr_docs=5` and each topic holds 6 unique documents. Without a
+    cap this returns 5.
+    """
+    docs = [f"topic zero doc {i}" for i in range(6)] + [f"topic one doc {i}" for i in range(6)]
+    topics_list = [0] * 6 + [1] * 6
+
+    model, c_tf_idf, documents, topics = minimal_topic_model(docs, topics_list)
+
+    repr_docs_mappings, _, _, repr_docs_ids = model._extract_representative_docs(
+        c_tf_idf,
+        documents,
+        topics,
+        nr_samples=2,
+        nr_repr_docs=5,
+    )
+
+    assert len(repr_docs_mappings[0]) == 2
+    assert len(repr_docs_mappings[1]) == 2
+
+    # Shuffling the frame before grouping must not leak documents across topics:
+    # every returned id has to belong to the topic it is reported under.
+    for topic, doc_ids in zip(sorted(topics.keys()), repr_docs_ids):
+        assert set(documents.loc[doc_ids, "Topic"]) == {topic}
+
+
+def test_with_diversity_no_duplicates(minimal_topic_model):
+    """MMR branch (diversity > 0) should also produce no duplicates."""
+    docs = [
+        "machine learning algorithms",
+        "deep learning neural networks",
+        "natural language processing",
+        "computer vision image analysis",
+        "data mining techniques",
+        "statistical modeling methods",
+    ]
+    topics_list = [0, 0, 0, 1, 1, 1]
+
+    model, c_tf_idf, documents, topics = minimal_topic_model(docs, topics_list)
+
+    repr_docs_mappings, _, _, _ = model._extract_representative_docs(
+        c_tf_idf,
+        documents,
+        topics,
+        nr_samples=500,
+        nr_repr_docs=3,
+        diversity=0.5,
+    )
+
+    for topic, docs_list in repr_docs_mappings.items():
+        assert len(docs_list) == len(set(docs_list)), f"Topic {topic} has duplicate representative docs with diversity"
+
+
+def test_multimodal_dedup_preserves_distinct_images(minimal_topic_model):
+    """Distinct images sharing identical captions must not collapse into one candidate.
+
+    Regression test for M01: `_extract_representative_docs` deduplicated on
+    `Document` text alone after dropping the `Image` column, so multiple images
+    captioned identically by an image-to-text model were treated as a single
+    candidate, starving `VisualRepresentation` of images below `nr_repr_images`.
+    """
+    docs = ["scenic view"] * 5 + ["street scene"] * 4
+    images = [f"img_{i}.jpg" for i in range(9)]
+    topics_list = [0] * 9
+
+    model, c_tf_idf, documents, topics = minimal_topic_model(docs, topics_list, images=images)
+
+    repr_docs_mappings, _, _, repr_docs_ids = model._extract_representative_docs(
+        c_tf_idf,
+        documents,
+        topics,
+        nr_samples=500,
+        nr_repr_docs=9,
+    )
+
+    # Only 2 distinct captions exist, but 9 distinct images back them. The buggy
+    # dedup collapsed this to 2 candidates total; the fix must keep all 9.
+    assert len(repr_docs_mappings[0]) == 9
+    assert len(repr_docs_ids[0]) == 9
+    assert len(set(repr_docs_ids[0])) == 9
+
+
+def test_diversity_with_duplicate_text_maps_correct_ids(minimal_topic_model):
+    """MMR branch must map indices positionally, not via a text-keyed lookup.
+
+    Regression test for the `doc_to_index` landmine noted in the PR review
+    (L04): once the dedup fix lets duplicate `Document` text survive dedup
+    (distinct images, same caption), a text-keyed reverse lookup collapses
+    every row sharing that text onto a single (wrong) index. `selected_indices`
+    must be computed positionally so `docs`/`repr_docs_ids` stay aligned.
+    """
+    docs = ["same caption"] * 4
+    images = ["img_0.jpg", "img_1.jpg", "img_2.jpg", "img_3.jpg"]
+    topics_list = [0, 0, 0, 0]
+
+    model, c_tf_idf, documents, topics = minimal_topic_model(docs, topics_list, images=images)
+
+    _, _, _, repr_docs_ids = model._extract_representative_docs(
+        c_tf_idf,
+        documents,
+        topics,
+        nr_samples=500,
+        nr_repr_docs=4,
+        diversity=0.5,
+    )
+
+    # All 4 rows share identical text; only the distinct `Image`/row identity
+    # tells them apart. A text-keyed `doc_to_index` lookup would map every
+    # returned document to the same (last-enumerated) index, collapsing ids.
+    assert sorted(repr_docs_ids[0]) == [0, 1, 2, 3]
+
+
+def test_multimodal_dedup_handles_unhashable_loaded_images(minimal_topic_model):
+    """Loaded (non-`str`) images must not crash `_extract_representative_docs`.
+
+    Regression test for the crash M01 introduced: `drop_duplicates` hashes its
+    subset columns, but `PIL.Image` sets `__hash__ = None`, so putting the
+    `Image` column directly into the dedup subset raises `TypeError:
+    unhashable type: 'Image'` the moment a pipeline carries loaded images
+    rather than string paths (e.g. the documented multimodal quickstart, which
+    loads a `datasets` image column directly). Distinct images sharing a
+    caption must still survive, and two images that are pixel-identical
+    (PIL's own `Image.__eq__`) must still collapse to one candidate.
+    """
+    Image = pytest.importorskip("PIL.Image")
+
+    distinct = [Image.new("RGB", (4, 4), color) for color in ("red", "green", "blue")]
+    duplicate_of_first = Image.new("RGB", (4, 4), "red")  # pixel-identical to distinct[0]
+
+    docs = ["same caption"] * 4
+    images = [*distinct, duplicate_of_first]
+    topics_list = [0, 0, 0, 0]
+
+    model, c_tf_idf, documents, topics = minimal_topic_model(docs, topics_list, images=images)
+
+    repr_docs_mappings, _, _, repr_docs_ids = model._extract_representative_docs(
+        c_tf_idf,
+        documents,
+        topics,
+        nr_samples=500,
+        nr_repr_docs=4,
+    )
+
+    # The pixel-identical duplicate collapses; the 3 distinct images survive.
+    assert len(repr_docs_mappings[0]) == 3
+    assert sorted(repr_docs_ids[0]) == [0, 1, 2]
diff --git a/tests/test_repr_docs_indexing.py b/tests/test_repr_docs_indexing.py
new file mode 100644
index 00000000..fa157cf1
--- /dev/null
+++ b/tests/test_repr_docs_indexing.py
@@ -0,0 +1,365 @@
+"""Tests for positional indexing in `_extract_representative_docs`.
+
+Verifies that representative documents map back to the correct topic when the
+same text appears in multiple topics, instead of matching by text membership.
+
+Run from BERTopic repo root:
+    pytest tests/test_repr_docs_indexing.py -v
+"""
+
+import pandas as pd
+import pytest
+from scipy.sparse import csr_matrix
+
+from bertopic import BERTopic
+
+
+def test_duplicate_text_across_topics(minimal_topic_model):
+    """Documents with identical text in different topics get correct doc_ids."""
+    # "shared text" appears in both topic 0 and topic 1
+    docs = [
+        "shared text",
+        "unique topic zero content",
+        "shared text",
+        "unique topic one content",
+    ]
+    topics_list = [0, 0, 1, 1]
+
+    model, c_tf_idf, documents, topics = minimal_topic_model(docs, topics_list)
+
+    _repr_docs_mappings, _repr_docs, _repr_docs_indices, repr_docs_ids = model._extract_representative_docs(
+        c_tf_idf, documents, topics, nr_samples=500, nr_repr_docs=3
+    )
+
+    # Verify each topic's representative doc_ids point to documents
+    # that actually belong to that topic
+    for topic_id, doc_ids in zip(sorted(topics.keys()), repr_docs_ids):
+        for doc_id in doc_ids:
+            actual_topic = documents.loc[doc_id, "Topic"]
+            assert actual_topic == topic_id, (
+                f"doc_id {doc_id} has topic {actual_topic} but was assigned as representative of topic {topic_id}"
+            )
+
+
+def test_all_identical_docs(minimal_topic_model):
+    """When all docs are identical, doc_ids should still be correct per topic.
+
+    All 3 docs per topic share the same text, so dedup collapses each topic
+    down to 1 candidate; `nr_repr_docs=2` can only return that 1. Asserting
+    the count pins the dedup interaction instead of looping over doc_ids
+    that a `[]` return would also satisfy.
+    """
+    docs = ["same text"] * 6
+    topics_list = [0, 0, 0, 1, 1, 1]
+
+    model, c_tf_idf, documents, topics = minimal_topic_model(docs, topics_list)
+
+    _repr_docs_mappings, _repr_docs, _repr_docs_indices, repr_docs_ids = model._extract_representative_docs(
+        c_tf_idf, documents, topics, nr_samples=500, nr_repr_docs=2
+    )
+
+    for topic_id, doc_ids in zip(sorted(topics.keys()), repr_docs_ids):
+        assert len(doc_ids) == 1, f"topic {topic_id}: expected exactly 1 doc_id after dedup, got {doc_ids}"
+        for doc_id in doc_ids:
+            actual_topic = documents.loc[doc_id, "Topic"]
+            assert actual_topic == topic_id, (
+                f"doc_id {doc_id} mapped to topic {actual_topic}, expected topic {topic_id}"
+            )
+
+
+def test_no_cross_topic_contamination(minimal_topic_model):
+    """Representative docs for a topic should not contain docs from another topic."""
+    docs = [
+        "alpha beta gamma",
+        "alpha beta delta",
+        "epsilon zeta eta",
+        "epsilon zeta theta",
+    ]
+    topics_list = [0, 0, 1, 1]
+
+    model, c_tf_idf, documents, topics = minimal_topic_model(docs, topics_list)
+
+    repr_docs_mappings, _, _, _repr_docs_ids = model._extract_representative_docs(
+        c_tf_idf, documents, topics, nr_samples=500, nr_repr_docs=2
+    )
+
+    for topic_id in topics.keys():
+        repr_doc_texts = repr_docs_mappings[topic_id]
+        topic_doc_texts = documents.loc[documents.Topic == topic_id, "Document"].tolist()
+        for doc in repr_doc_texts:
+            assert doc in topic_doc_texts, (
+                f"Representative doc '{doc}' for topic {topic_id} not found in that topic's documents"
+            )
+
+
+def test_doc_ids_count_matches_nr_repr_docs(minimal_topic_model):
+    """doc_ids count should match nr_repr_docs per topic."""
+    docs = [
+        "doc alpha one",
+        "doc beta two",
+        "doc gamma three",
+        "doc delta four",
+        "doc epsilon five",
+        "doc zeta six",
+    ]
+    topics_list = [0, 0, 0, 1, 1, 1]
+
+    model, c_tf_idf, documents, topics = minimal_topic_model(docs, topics_list)
+
+    _, _, _, repr_docs_ids = model._extract_representative_docs(
+        c_tf_idf, documents, topics, nr_samples=500, nr_repr_docs=2
+    )
+
+    for topic_id, doc_ids in zip(sorted(topics.keys()), repr_docs_ids):
+        assert len(doc_ids) == 2, f"Topic {topic_id} should have 2 doc_ids, got {len(doc_ids)}"
+
+
+def test_doc_ids_are_valid_dataframe_indices(minimal_topic_model):
+    """All returned doc_ids should be valid indices into the original DataFrame."""
+    docs = [
+        "shared text",
+        "unique topic zero content",
+        "shared text",
+        "unique topic one content",
+        "more topic one docs",
+    ]
+    topics_list = [0, 0, 1, 1, 1]
+
+    model, c_tf_idf, documents, topics = minimal_topic_model(docs, topics_list)
+
+    _, _, _, repr_docs_ids = model._extract_representative_docs(
+        c_tf_idf, documents, topics, nr_samples=500, nr_repr_docs=3
+    )
+
+    valid_indices = set(documents.index.tolist())
+    for doc_ids in repr_docs_ids:
+        for doc_id in doc_ids:
+            assert doc_id in valid_indices, f"doc_id {doc_id} not in DataFrame index"
+
+
+def test_duplicate_text_with_diversity(minimal_topic_model):
+    """MMR branch should also map doc_ids correctly with duplicate text."""
+    docs = [
+        "machine learning algorithms applied",
+        "machine learning methods used",
+        "natural language processing tasks",
+        "natural language understanding models",
+    ]
+    topics_list = [0, 0, 1, 1]
+
+    model, c_tf_idf, documents, topics = minimal_topic_model(docs, topics_list)
+
+    _, _, _, repr_docs_ids = model._extract_representative_docs(
+        c_tf_idf,
+        documents,
+        topics,
+        nr_samples=500,
+        nr_repr_docs=2,
+        diversity=0.5,
+    )
+
+    for topic_id, doc_ids in zip(sorted(topics.keys()), repr_docs_ids):
+        for doc_id in doc_ids:
+            actual_topic = documents.loc[doc_id, "Topic"]
+            assert actual_topic == topic_id, (
+                f"doc_id {doc_id} has topic {actual_topic} but assigned to topic {topic_id}"
+            )
+
+    for topic_id, doc_ids in zip(sorted(topics.keys()), repr_docs_ids):
+        # With positional indexing, doc_ids count should equal nr_repr_docs
+        # (or fewer if topic has fewer docs)
+        assert len(doc_ids) == 2, f"Topic {topic_id} should have 2 doc_ids, got {len(doc_ids)}"
+
+
+@pytest.mark.parametrize("diversity", [None, 0.5])
+def test_doc_ids_are_index_labels_not_positions(minimal_topic_model, diversity):
+    """`doc_ids` must be DataFrame index labels, not positions into `documents`.
+
+    Uses a non-contiguous, shifted index and an `ID` column deliberately distinct
+    from the index (mirroring the zero-shot path where `ID` is reset to
+    `range(len(documents))` independently of the original index labels, see
+    `_bertopic.py`'s zero-shot handling). If a regression returned positions
+    instead of labels, this test would catch it even though a default
+    RangeIndex-based test could not (label == position there).
+    """
+    docs = [
+        "doc alpha one",
+        "doc beta two",
+        "doc gamma three",
+        "doc delta four",
+        "doc epsilon five",
+        "doc zeta six",
+    ]
+    topics_list = [0, 0, 0, 1, 1, 1]
+    shifted_index = [100, 101, 102, 103, 104, 105]
+    # ID intentionally different from both index and position
+    ids = [900, 901, 902, 903, 904, 905]
+
+    model, c_tf_idf, documents, topics = minimal_topic_model(docs, topics_list, index=shifted_index, ids=ids)
+
+    _, _, _, repr_docs_ids = model._extract_representative_docs(
+        c_tf_idf,
+        documents,
+        topics,
+        nr_samples=500,
+        nr_repr_docs=2,
+        diversity=diversity,
+    )
+
+    valid_labels = set(documents.index.tolist())
+    for topic_id, doc_ids in zip(sorted(topics.keys()), repr_docs_ids):
+        for doc_id in doc_ids:
+            assert doc_id in valid_labels, f"doc_id {doc_id} is not a valid index label"
+            assert doc_id not in range(len(docs)), (
+                f"doc_id {doc_id} looks like a position (0..{len(docs) - 1}), not a shifted index label"
+            )
+            assert documents.loc[doc_id, "Topic"] == topic_id, (
+                f"doc_id {doc_id} has topic {documents.loc[doc_id, 'Topic']}, expected {topic_id}"
+            )
+
+
+@pytest.mark.parametrize("diversity", [None, 0.5])
+def test_unsorted_topics_keys_map_docs_to_correct_topic(minimal_topic_model, diversity):
+    """`repr_docs_mappings` must attach documents to the correct topic even when
+    the `topics` dict's key insertion order is not sorted.
+
+    The extraction loop iterates `sorted(topics.keys())` (see `_bertopic.py`,
+    `labels = sorted(list(topics.keys()))`), so `repr_docs`/`repr_docs_indices`
+    are built in sorted order. If `repr_docs_mappings` were instead built by
+    zipping against `topics.keys()` in its original (unsorted) insertion order,
+    documents would be attached to the wrong topic.
+    """
+    docs = [
+        "alpha beta gamma",
+        "alpha beta delta",
+        "epsilon zeta eta",
+        "epsilon zeta theta",
+    ]
+    topics_list = [0, 0, 1, 1]
+
+    # Reversed insertion order: sorted order is [0, 1], insertion order is [1, 0]
+    model, c_tf_idf, documents, topics = minimal_topic_model(docs, topics_list, topic_order=[1, 0])
+    assert list(topics.keys()) == [1, 0]
+
+    repr_docs_mappings, _, _, _ = model._extract_representative_docs(
+        c_tf_idf,
+        documents,
+        topics,
+        nr_samples=500,
+        nr_repr_docs=2,
+        diversity=diversity,
+    )
+
+    for topic_id in topics.keys():
+        repr_doc_texts = repr_docs_mappings[topic_id]
+        topic_doc_texts = documents.loc[documents.Topic == topic_id, "Document"].tolist()
+        for doc in repr_doc_texts:
+            assert doc in topic_doc_texts, (
+                f"Representative doc '{doc}' for topic {topic_id} not found in that topic's "
+                f"documents (topics dict insertion order was {list(topics.keys())})"
+            )
+
+
+@pytest.mark.parametrize("diversity", [None, 0.5])
+@pytest.mark.parametrize("topic_order", [[0, 1], [1, 0]])
+def test_mappings_agree_with_repr_docs_ids(minimal_topic_model, diversity, topic_order):
+    """`repr_docs_mappings[t]` texts must correspond to the same documents as
+    `repr_docs_ids` for topic `t`, regardless of the `topics` dict key order.
+    """
+    docs = [
+        "doc alpha one",
+        "doc beta two",
+        "doc gamma three",
+        "doc delta four",
+        "doc epsilon five",
+        "doc zeta six",
+    ]
+    topics_list = [0, 0, 0, 1, 1, 1]
+
+    model, c_tf_idf, documents, topics = minimal_topic_model(docs, topics_list, topic_order=topic_order)
+
+    repr_docs_mappings, _, _, repr_docs_ids = model._extract_representative_docs(
+        c_tf_idf,
+        documents,
+        topics,
+        nr_samples=500,
+        nr_repr_docs=2,
+        diversity=diversity,
+    )
+
+    # repr_docs_ids is built in sorted-label order regardless of topics dict order
+    for topic_id, doc_ids in zip(sorted(topics.keys()), repr_docs_ids):
+        expected_texts = set(documents.loc[doc_ids, "Document"].tolist())
+        actual_texts = set(repr_docs_mappings[topic_id])
+        assert actual_texts == expected_texts, (
+            f"topic {topic_id} (topic_order={topic_order}): mappings {actual_texts} "
+            f"do not match repr_docs_ids-derived texts {expected_texts}"
+        )
+
+
+@pytest.mark.parametrize(
+    "documents",
+    [
+        pytest.param(pd.DataFrame({"Document": [], "ID": [], "Topic": []}), id="empty_documents"),
+        pytest.param(
+            pd.DataFrame({"Document": ["doc one", "doc two"], "ID": [0, 1], "Topic": [None, None]}),
+            id="all_nan_topic",
+        ),
+    ],
+)
+def test_extract_representative_docs_raises_clear_error_on_no_valid_topics(documents):
+    """An empty `documents` or an all-NaN `Topic` column (topics not assigned yet) both make
+    `groupby("Topic")` drop every group, which previously reached `pd.concat([])` and raised
+    pandas's opaque `ValueError: No objects to concatenate`. A guard must raise a clear error
+    instead, before that point.
+    """
+    model = BERTopic()
+
+    with pytest.raises(ValueError, match="No documents with a valid `Topic` assignment"):
+        model._extract_representative_docs(c_tf_idf=csr_matrix((0, 0)), documents=documents, topics={})
+
+
+def test_unequal_topic_sizes_offset_arithmetic(minimal_topic_model):
+    """`repr_docs_indices` offset arithmetic
+    (`repr_docs_indices[-1][-1] + i + 1 if index != 0 else i`) must produce a contiguous,
+    gap-free, non-overlapping partition of `repr_docs` even when topics have unequal sizes.
+    Every other test in this suite uses equal per-topic counts, where an off-by-one in the
+    offset cannot show up.
+    """
+    docs = [
+        "topic zero solo doc",
+        "topic one doc a",
+        "topic one doc b",
+        "topic one doc c",
+        "topic two doc a",
+        "topic two doc b",
+    ]
+    topics_list = [0, 1, 1, 1, 2, 2]
+
+    model, c_tf_idf, documents, topics = minimal_topic_model(docs, topics_list)
+
+    repr_docs_mappings, repr_docs, repr_docs_indices, repr_docs_ids = model._extract_representative_docs(
+        c_tf_idf, documents, topics, nr_samples=500, nr_repr_docs=3
+    )
+
+    expected_counts = {0: 1, 1: 3, 2: 2}
+    for topic_id, doc_ids in zip(sorted(topics.keys()), repr_docs_ids):
+        assert len(doc_ids) == expected_counts[topic_id], (
+            f"topic {topic_id}: expected {expected_counts[topic_id]} doc_ids, got {len(doc_ids)}"
+        )
+
+    # Contiguous, gap-free, non-overlapping: each topic's first index must pick up exactly
+    # where the previous topic's last index left off. An off-by-one in the offset arithmetic
+    # would either skip an index (gap) or repeat one (overlap), and this is where it would show.
+    flat_indices = [i for indices in repr_docs_indices for i in indices]
+    assert flat_indices == list(range(len(repr_docs))), (
+        f"repr_docs_indices is not a contiguous partition of range(len(repr_docs)): {flat_indices}"
+    )
+
+    # repr_docs_mappings slices must correspond to the same documents as repr_docs_ids.
+    for topic_id, doc_ids in zip(sorted(topics.keys()), repr_docs_ids):
+        expected_texts = set(documents.loc[doc_ids, "Document"].tolist())
+        actual_texts = set(repr_docs_mappings[topic_id])
+        assert actual_texts == expected_texts, (
+            f"topic {topic_id}: mappings {actual_texts} do not match repr_docs_ids-derived texts {expected_texts}"
+        )
diff --git a/tests/test_representation/test_visual.py b/tests/test_representation/test_visual.py
new file mode 100644
index 00000000..8ae47eec
--- /dev/null
+++ b/tests/test_representation/test_visual.py
@@ -0,0 +1,73 @@
+import pytest
+
+PIL = pytest.importorskip("PIL")
+from PIL import Image  # noqa: E402
+
+from bertopic.representation import _visual as visual_module  # noqa: E402
+from bertopic.representation import VisualRepresentation  # noqa: E402
+
+
+def test_extract_topics_indexes_images_by_label_not_position(minimal_topic_model, monkeypatch):
+    """`_extract_representative_docs` returns index labels (not positions) in
+    `repr_docs_ids`. `VisualRepresentation.extract_topics` must look images up by
+    those labels; using a non-contiguous/shifted DataFrame index catches any
+    accidental positional lookup.
+    """
+    docs = [
+        "doc alpha one",
+        "doc beta two",
+        "doc gamma three",
+        "doc delta four",
+        "doc epsilon five",
+        "doc zeta six",
+    ]
+    topics_list = [0, 0, 0, 1, 1, 1]
+    shifted_index = [50, 51, 52, 53, 54, 55]
+
+    model, c_tf_idf, documents, topics = minimal_topic_model(docs, topics_list, index=shifted_index)
+    # `_outliers` inspects `topic_sizes_` to know whether topic -1 is present.
+    model.topic_sizes_ = {0: 3, 1: 3}
+
+    # Attach a distinctive, non-string "Image" per document, keyed by its
+    # (shifted) index label so we can verify which document each captured
+    # image actually corresponds to.
+    images_by_label = {}
+    for label in documents.index:
+        image = Image.new("RGB", (10, 10))
+        image.info["label"] = label
+        images_by_label[label] = image
+    documents = documents.copy()
+    documents["Image"] = [images_by_label[label] for label in documents.index]
+
+    captured_images_to_combine = {}
+
+    def fake_get_concat_tile_resize(im_list_2d, image_height=600, image_squares=False):
+        # Flatten the 2D grid of images and remember which labels were passed in.
+        captured_images_to_combine[current_topic[0]] = [image.info["label"] for row in im_list_2d for image in row]
+        return Image.new("RGB", (10, 10))
+
+    monkeypatch.setattr(visual_module, "get_concat_tile_resize", fake_get_concat_tile_resize)
+
+    # `extract_topics` doesn't expose which topic is currently being processed
+    # to `get_concat_tile_resize`, so track it via the tqdm loop order, which is
+    # `sorted(topics.keys())` (see `_visual.py`).
+    current_topic = [None]
+
+    def fake_tqdm(iterable, *args, **kwargs):
+        for item in iterable:
+            current_topic[0] = item
+            yield item
+
+    monkeypatch.setattr(visual_module, "tqdm", fake_tqdm)
+
+    representation_model = VisualRepresentation(nr_repr_images=3, nr_samples=500)
+    representation_model.extract_topics(model, documents, c_tf_idf, topics)
+
+    assert set(captured_images_to_combine.keys()) == {0, 1}
+    for topic_id, labels in captured_images_to_combine.items():
+        assert labels, f"no images captured for topic {topic_id}"
+        for label in labels:
+            assert documents.loc[label, "Topic"] == topic_id, (
+                f"image with label {label} (topic {documents.loc[label, 'Topic']}) leaked into "
+                f"topic {topic_id}'s collage"
+            )