diff --git a/bertopic/_bertopic.py b/bertopic/_bertopic.py
index d53e7f95..a1c721b6 100644
--- a/bertopic/_bertopic.py
+++ b/bertopic/_bertopic.py
@@ -667,8 +667,8 @@ def transform(
             logger.info("Cluster - Completed \u2713")
 
             # Map probabilities and predictions
-            predictions = self._topics.map_predictions(predictions, from_original=True)
-            probabilities = self._topics.map_probabilities(probabilities, from_original=True)
+            predictions = self._topics.map_predictions(predictions)
+            probabilities = self._topics.align_probabilities(probabilities)
 
         return predictions, probabilities
 
@@ -776,13 +776,11 @@ def partial_fit(
             for topic_id in new_topic_ids:
                 self._topics.topics[topic_id] = Topic(id=topic_id, nr_documents=0)
 
-            self._topics._original_predictions = np.concatenate(
-                [self._topics._original_predictions, np.array(corpus.topics)]
-            )
+            self._topics.predictions.extend(int(topic) for topic in corpus.topics)
 
-            if corpus.probabilities is not None:
-                self._topics._original_probabilities = np.concatenate(
-                    [self._topics._original_probabilities, corpus.probabilities]
+            if corpus.probabilities is not None and self._topics.probabilities is not None:
+                self._topics.probabilities = np.concatenate(
+                    [self._topics.probabilities, corpus.probabilities]
                 )
 
             for topic_id in set(corpus.topics):
@@ -1235,9 +1233,12 @@ def get_document_info(
             if len(self.probabilities_.shape) == 1:
                 data["Probability"] = self.probabilities_.tolist()
             else:
+                # Column `j` holds topic `topic_ids()[j]`, so the assigned topic's
+                # probability is read directly rather than derived
+                topic_ids = self._topics.topic_ids()
                 data["Probability"] = [
-                    max(probs) if tid != -1 else 1 - sum(probs)
-                    for tid, probs in zip(predictions, self.probabilities_)
+                    float(probabilities[topic_ids.index(topic_id)])
+                    for topic_id, probabilities in zip(predictions, self.probabilities_)
                 ]
 
         # Custom metadata
@@ -1600,9 +1601,8 @@ def merge_topics(
         self._topics.merge(mapping)
 
         # Map corpus topics to match merged state and then sort by frequency
-        corpus.map_topics_and_probabilities(self._topics, from_original=False)
         self._topics.sort_by_frequency()
-        corpus.map_topics_and_probabilities(self._topics, from_original=False)
+        corpus.topics, corpus.probabilities = self._topics.predictions, self._topics.probabilities
 
         # Recalculate representations from merged documents
         self._extract_representations(corpus)
@@ -2490,27 +2490,25 @@ def _cluster_embeddings(self, corpus: Corpus, partial_fit: bool = False) -> Corp
             except AttributeError:
                 corpus.topics = corpus.y
 
-        # Create Topics object and sort by frequency
-        if not partial_fit:
-            self._topics.initialize(corpus.topics).sort_by_frequency()
-            corpus.map_topics_and_probabilities(self._topics, from_original=True)
-
-        # Extract probabilities
+        # Extract probabilities in the cluster model's own label order
         if hasattr(self.hdbscan_model, "probabilities_"):
             corpus.probabilities = self.hdbscan_model.probabilities_
 
             if self.calculate_probabilities and is_supported_hdbscan(self.hdbscan_model):
                 corpus.probabilities = hdbscan_delegator(self.hdbscan_model, "all_points_membership_vectors")
 
-                # HDBSCAN only produces probabilities for non-outliers. The outliers
-                # get a probability of 1 - sum(probabilities). Update the `corpus.probabilities` to
-                # add this new column at the beginning of the array.
+                # HDBSCAN gives a column per cluster and none for outliers, whose share is
+                # therefore whatever is left over. Column 0 then lines up with topic -1.
                 if -1 in corpus.topics:
-                    outlier_probs = 1 - np.sum(corpus.probabilities, axis=1)
-                    corpus.probabilities = np.hstack([outlier_probs.reshape(-1, 1), corpus.probabilities])
+                    outlier_probabilities = 1 - np.sum(corpus.probabilities, axis=1)
+                    corpus.probabilities = np.hstack(
+                        [outlier_probabilities.reshape(-1, 1), corpus.probabilities]
+                    )
 
-            if not partial_fit:
-                self._topics._original_probabilities = corpus.probabilities.copy()
+        # Hand the rows to Topics, which owns them from here on, and renumber by frequency
+        if not partial_fit:
+            self._topics.initialize(corpus.topics, probabilities=corpus.probabilities).sort_by_frequency()
+            corpus.topics, corpus.probabilities = self._topics.predictions, self._topics.probabilities
 
         logger.info("Cluster - Completed \u2713")
 
@@ -2992,11 +2990,10 @@ def _reduce_to_n_topics(self, corpus: Corpus, use_ctfidf: bool = False) -> Corpu
             for topic_id in self._topics.topic_ids()
         }
         self._topics.merge(mappings)
-        corpus.map_topics_and_probabilities(self._topics, from_original=False)
 
-        # Update frequency
+        # Update frequency, then take the rows back from the single store that owns them
         self._topics.sort_by_frequency()
-        corpus.map_topics_and_probabilities(self._topics, from_original=False)
+        corpus.topics, corpus.probabilities = self._topics.predictions, self._topics.probabilities
 
         # Recalculate topic representations
         self._extract_representations(corpus=corpus, verbose=self.verbose)
@@ -3057,11 +3054,10 @@ def _auto_reduce_topics(self, corpus: Corpus, use_ctfidf: bool = False) -> Corpu
             for topic_id in self._topics.topic_ids()
         }
         self._topics.merge(mappings)
-        corpus.map_topics_and_probabilities(self._topics, from_original=False)
 
-        # Update frequency
+        # Update frequency, then take the rows back from the single store that owns them
         self._topics.sort_by_frequency()
-        corpus.map_topics_and_probabilities(self._topics, from_original=False)
+        corpus.topics, corpus.probabilities = self._topics.predictions, self._topics.probabilities
 
         # Recalculate topic representations
         self._extract_representations(corpus=corpus, verbose=self.verbose)
diff --git a/bertopic/_corpus.py b/bertopic/_corpus.py
index fe3d410e..37f3c2ee 100644
--- a/bertopic/_corpus.py
+++ b/bertopic/_corpus.py
@@ -4,8 +4,6 @@
 from scipy.sparse import csr_matrix
 from collections import defaultdict
 
-from bertopic._topics import Topics
-
 
 class Modality(str, Enum):
     """What a row is, which determines whether its source lives in `documents` or `media`."""
@@ -157,80 +155,6 @@ def average_embeddings_by_topic(self) -> dict[int, np.ndarray]:
         averaged_sorted = dict(sorted(averaged.items()))
         return averaged_sorted
 
-    def map_topics_and_probabilities(self, topics: Topics, from_original: bool = False) -> None:
-        """Map both topics and probabilities to the reduced topics using the provided Topics object.
-
-        Arguments:
-            topics: A Topics object containing the mapping information.
-            from_original: Whether to map from the original topics to the current ones.
-        """
-        self.map_topics(topics, from_original=from_original)
-
-        # Only map probabilities if they are 2-dimensional since only then they
-        # correspond to topic probabilities which might be reduced or reordered.
-        if self.probabilities is not None:
-            if len(self.probabilities.shape) > 1:
-                self.map_probabilities(topics, from_original=from_original)
-
-    def map_topics(self, topics: Topics, from_original: bool = False) -> None:
-        """Map the topics to the reduced topics using the provided Topics object.
-
-        Arguments:
-            topics: A Topics object containing the mapping information.
-            from_original: Whether to map from the original topics to the current ones.
-        """
-        self.topics = [
-            topics.mapping.map(prediction, from_original=from_original) for prediction in self.topics
-        ]
-
-    def map_probabilities(self, topics: Topics, from_original: bool = False) -> None:
-        """Map the (2-dimensional) probabilities to the reduced topics.
-
-        There are two scenarios based on the mappings in the Topics object:
-        * The order of topics has changed (e.g., after sorting by frequency).
-          In this case, the probabilities are simply reordered.
-        * Some topics have been merged. In this case, the probabilities
-          of the merged topics are summed together and assigned to the new topic.
-
-        Note that the outlier topic (-1), if present, is skipped during this process.
-        If it is present, it is always at the zero-th index of the initial probabilities matrix
-        and should remain so after mapping.
-
-        Arguments:
-            topics: A Topics object containing the mapping information.
-            from_original: If True, mappings are obtained from the original topics.
-                           If False, mappings are obtained from the most recent topics.
-        """
-        # Check scenario based on mappings
-        mappings = topics.get_mappings(from_original=from_original)
-
-        # Scenario 1: Reordering
-        if len(set(mappings.values())) == len(mappings):
-            nr_topics = len(set(mappings.values()))
-            new_order = [0] * nr_topics
-            for old_topic, new_topic in mappings.items():
-                if old_topic == -1:
-                    continue  # Skip outlier topic
-                new_topic_idx = new_topic + self._outliers
-                old_topic_idx = old_topic + self._outliers
-                new_order[new_topic_idx] = old_topic_idx
-
-            self.probabilities = self.probabilities[:, new_order]
-
-        # Scenario 2: Merging
-        else:
-            nr_new_topics = len(set(mappings.values()))
-            new_probabilities = np.zeros((self.probabilities.shape[0], nr_new_topics))
-
-            for old_topic, new_topic in mappings.items():
-                if old_topic == -1:
-                    continue  # Skip outlier topic
-                new_topic_idx = new_topic + self._outliers
-                old_topic_idx = old_topic + self._outliers
-                new_probabilities[:, new_topic_idx] += self.probabilities[:, old_topic_idx]
-
-            self.probabilities = new_probabilities
-
     def sort_topics_by_frequency(self) -> "Corpus":
         """Maps the label of each topic to its frequency rank with
         the outlier topic (-1) always being the -1 topic.
diff --git a/bertopic/_save_utils.py b/bertopic/_save_utils.py
index d4d3db72..624d15c8 100644
--- a/bertopic/_save_utils.py
+++ b/bertopic/_save_utils.py
@@ -129,7 +129,7 @@ def migrate_topics_pre_0_17_4(topics_dict: dict) -> Topics:
 
     # Build Topics object
     topics = Topics()
-    topics._original_predictions = np.array(topics_dict.get("topics", []))
+    topics.predictions = list(topics_dict.get("topics", []))
     topics.actions = [TopicAction.INITIALIZED]
 
     # Reconstruct mapping from topic_mapper (2D array of mappings history)
diff --git a/bertopic/_topics.py b/bertopic/_topics.py
index 5ca912b2..2720cf02 100644
--- a/bertopic/_topics.py
+++ b/bertopic/_topics.py
@@ -202,93 +202,53 @@ class TopicType(str, Enum):
 
 @dataclass
 class TopicMapping:
-    """Tracks cumulative topic ID transformations from original to current state.
-    Also keeps track of the most recent mapping applied.
+    """Tracks how a cluster model's original labels map to the current topic IDs.
+
+    Only `transform` needs this. A cluster model hands back the labels it invented while
+    fitting, which have to be translated into whatever the topics were renumbered to
+    since. Rows that were part of the fit keep their assignments in `Topics.predictions`
+    instead, so nothing here describes the fitted corpus.
     """
 
     _mapping: dict[int, int] = field(default_factory=dict)
-    _recent_mapping: dict[int, int] = field(default_factory=dict)
 
-    def apply(self, new_mapping: dict[int, int]) -> None:
-        """Compose a new mapping: original -> current becomes original -> new_current."""
+    def apply(self, old_to_new: dict[int, int]) -> None:
+        """Compose a renumbering, keeping the mapping original -> current."""
         if not self._mapping:
-            self._mapping = new_mapping.copy()
-            self._recent_mapping = new_mapping.copy()
-        else:
-            missing = sorted(set(self._mapping.values()) - set(new_mapping))
-            if missing:
-                raise ValueError(
-                    f"The mapping is missing topics {missing}. Every current topic must appear "
-                    "in a new mapping, including topics that hold no documents."
-                )
-            self._mapping = {original: new_mapping[current] for original, current in self._mapping.items()}
-            self._recent_mapping = new_mapping.copy()
-
-    def map(self, topic_id: int, from_original: bool = True) -> int:
-        """Map an ID to its current ID.
+            self._mapping = old_to_new.copy()
+            return
+
+        missing = sorted(set(self._mapping.values()) - set(old_to_new))
+        if missing:
+            raise ValueError(
+                f"The mapping is missing topics {missing}. Every current topic must appear "
+                "in a new mapping, including topics that hold no documents."
+            )
+        self._mapping = {original: old_to_new[current] for original, current in self._mapping.items()}
 
-        Arguments:
-            topic_id: The topic ID to map.
-            from_original: If True, map from the original ID to current.
-                           If False, map from the last applied mapping to current.
-        """
-        if from_original:
-            return self._mapping.get(topic_id, topic_id)
-        else:
-            return self._recent_mapping.get(topic_id, topic_id)
-
-    def map_predictions(self, predictions: list[int], from_original: bool = True) -> list[int]:
-        """Map a list of original predictions to current IDs."""
-        return [self.map(p, from_original) for p in predictions]
-
-    def map_probabilities(self, probabilities: np.ndarray, from_original: bool = True) -> np.ndarray:
-        """Map a 2D array of probabilities to current IDs."""
-        if from_original:
-            mapped_probs = np.zeros((probabilities.shape[0], len(set(self._mapping.values()))))
-            for original_id, current_id in self._mapping.items():
-                if original_id >= 0 and current_id >= 0:
-                    mapped_probs[:, current_id] = probabilities[:, original_id]
-            return mapped_probs
-        else:
-            mapped_probs = np.zeros((probabilities.shape[0], len(set(self._recent_mapping.values()))))
-            for last_id, current_id in self._recent_mapping.items():
-                if last_id >= 0 and current_id >= 0:
-                    mapped_probs[:, current_id] = probabilities[:, last_id]
-            return mapped_probs
+    def map(self, topic_id: int) -> int:
+        """Translate one original cluster label into its current topic ID."""
+        return self._mapping.get(topic_id, topic_id)
 
-    def add_new_topics(self, new_mappings: dict[int, int]) -> None:
-        """Add mappings for newly discovered topics in online learning.
-
-        Arguments:
-            new_mappings: Mapping from new cluster IDs to new topic IDs
-        """
-        self._mapping.update(new_mappings)
-        self._recent_mapping.update(new_mappings)
+    def map_predictions(self, predictions: list[int] | np.ndarray) -> list[int]:
+        """Translate a cluster model's labels into current topic IDs."""
+        return [self.map(int(prediction)) for prediction in predictions]
 
     def reset(self) -> None:
-        """Clear the mapping (e.g., after re-fitting)."""
+        """Clear the mapping, so current IDs become the originals."""
         self._mapping.clear()
-        self._recent_mapping.clear()
 
     def to_dict(self) -> dict:
         """Serialize to dictionary."""
-        return {
-            "mapping": {str(k): v for k, v in self._mapping.items()},
-            "recent_mapping": {str(k): v for k, v in self._recent_mapping.items()},
-        }
+        return {"mapping": {str(original): current for original, current in self._mapping.items()}}
 
     @classmethod
     def from_dict(cls, data: dict) -> "TopicMapping":
         """Deserialize from dictionary."""
         mapping = cls()
-        mapping._mapping = {int(k): v for k, v in data.get("mapping", {}).items()}
-        mapping._recent_mapping = {int(k): v for k, v in data.get("recent_mapping", {}).items()}
+        mapping._mapping = {int(original): current for original, current in data.get("mapping", {}).items()}
         return mapping
 
-    def copy(self) -> "TopicMapping":
-        """Create a copy of this mapping."""
-        return TopicMapping.from_dict(self.to_dict())
-
 
 @dataclass
 class Topic:
@@ -486,14 +446,10 @@ class Topics:
     topics: dict[int, Topic] = field(default_factory=dict)
     mapping: TopicMapping = field(default_factory=TopicMapping)
 
-    # Document metadata (optional)
-    # NOTE: These are the original predictions/probabilities from the
-    # clustering algorithm before any remapping.
-    _original_predictions: np.ndarray = field(default_factory=lambda: np.array([]))
-    _original_probabilities: np.ndarray | None = None
-
-    # Zero-shot probabilities (optional)
-    _zeroshot_probabilities: np.ndarray | None = None
+    # One assignment per row, always in current topic IDs. Rows are whatever was fitted,
+    # which is not necessarily the user's documents.
+    predictions: list[int] = field(default_factory=list)
+    probabilities: np.ndarray | None = None
 
     # History of actions applied to this collection
     actions: list[TopicAction] = field(default_factory=list)
@@ -508,22 +464,6 @@ def labels(self) -> dict[int, str]:
         labels = {topic.id: topic.label if topic.label is not None else "" for topic in self.topics.values()}
         return dict(sorted(labels.items()))
 
-    @property
-    def predictions(self) -> list[int]:
-        """Get current predictions (mapped from original)."""
-        return self.mapping.map_predictions(self._original_predictions.tolist(), from_original=True)
-
-    @property
-    def probabilities(self) -> np.ndarray | None:
-        """Get current probabilities."""
-        if self._zeroshot_probabilities is not None:
-            return self._zeroshot_probabilities
-        elif self._original_probabilities is not None:
-            if self._original_probabilities.ndim == 1:
-                return self._original_probabilities
-            elif self._original_probabilities.ndim == 2:
-                return self.mapping.map_probabilities(self._original_probabilities, from_original=True)
-
     @property
     def c_tf_idf(self) -> csr_matrix:
         """Get c-TF-IDF matrix for all topics in ascending topic ID order
@@ -587,8 +527,9 @@ def initialize(
         predictions: list[int] | np.ndarray = None,
         zeroshot_labels: list[str] | None = None,
         topic_type: TopicType = TopicType.NORMAL,
+        probabilities: np.ndarray | None = None,
     ):
-        """Initialize topics from clustering predictions."""
+        """Initialize topics from a cluster model's assignments."""
         if isinstance(predictions, np.ndarray):
             predictions = predictions.tolist()
 
@@ -608,8 +549,8 @@ def initialize(
                 _label=label,
             )
 
-        # Set original predictions/probabilities
-        self._original_predictions = np.array(predictions)
+        self.predictions = list(predictions)
+        self.probabilities = probabilities
 
         # Log the initialization
         self.add_action(TopicAction.INITIALIZED)
@@ -694,13 +635,33 @@ def get_topics(self, topic_ids: list[int]) -> "Topics":
             actions=self.actions.copy(),
         )
 
+    def _apply_to_rows(self, old_to_new: dict[int, int], column_order: list[int]) -> None:
+        """Move every row's assignment and probability column to the new topic IDs.
+
+        Columns are summed rather than overwritten, so this covers merging and deleting
+        as well as renumbering: topics that collapse into one add their mass together,
+        and deleted topics map to -1 and so add theirs to the outlier.
+        """
+        self.predictions = [old_to_new[prediction] for prediction in self.predictions]
+
+        if self.probabilities is None or self.probabilities.ndim != 2:
+            return
+
+        new_ids = sorted(set(old_to_new.values()))
+        remapped = np.zeros((self.probabilities.shape[0], len(new_ids)))
+        for position, old_id in enumerate(column_order):
+            remapped[:, new_ids.index(old_to_new[old_id])] += self.probabilities[:, position]
+        self.probabilities = remapped
+
     def remap(self, old_to_new: dict[int, int]) -> None:
         """Apply an ID remapping to topics and update cumulative mapping.
         This is expected to be a one to one mapping.
         """
+        column_order = sorted(self.topics)
         for topic in self.topics.values():
             topic.id = old_to_new[topic.id]
         self.topics = {topic.id: topic for topic in self.topics.values()}
+        self._apply_to_rows(old_to_new, column_order)
         self.mapping.apply(old_to_new)
 
     def merge(self, old_to_new: dict[int, int]) -> None:
@@ -721,6 +682,8 @@ def merge(self, old_to_new: dict[int, int]) -> None:
             old_to_new: A dictionary mapping old topic IDs to new topic IDs.
                         New topics are fewer than old topics.
         """
+        column_order = sorted(self.topics)
+
         # Group old topic IDs by their new target ID
         new_to_old: dict[int, list[int]] = defaultdict(list)
         for old_id, new_id in old_to_new.items():
@@ -773,6 +736,7 @@ def merge(self, old_to_new: dict[int, int]) -> None:
             )
 
         self.topics = merged_topics
+        self._apply_to_rows(old_to_new, column_order)
         self.mapping.apply(old_to_new)
         self.add_action(TopicAction.MERGED)
 
@@ -828,6 +792,7 @@ def delete(self, topics: list[int] | int) -> None:
             self.topics[-1].nr_documents += deleted_doc_count
 
         # Build mapping: deleted -> -1, others -> themselves
+        column_order = sorted(self.topics)
         old_to_new = {topic_id: -1 if topic_id in topics else topic_id for topic_id in self.topics.keys()}
         old_to_new[-1] = -1
 
@@ -836,43 +801,38 @@ def delete(self, topics: list[int] | int) -> None:
             if topic_id in self.topics:
                 del self.topics[topic_id]
 
+        self._apply_to_rows(old_to_new, column_order)
         self.mapping.apply(old_to_new)
         self.add_action(TopicAction.DELETED)
 
-    def map_predictions(self, predictions: list[int], from_original: bool) -> list[int]:
-        """Map a list of original predictions to current IDs.
-
-        Arguments:
-            predictions: List of topic IDs to map.
-            from_original: If True, map from original IDs to current.
-                           If False, map from last applied mapping to current.
-        """
-        return [int(self.mapping.map(prediction, from_original=from_original)) for prediction in predictions]
+    def map_predictions(self, predictions: list[int] | np.ndarray) -> list[int]:
+        """Translate a cluster model's labels into current topic IDs, for new rows."""
+        return self.mapping.map_predictions(predictions)
 
-    def map_probabilities(self, probabilities: np.ndarray, from_original: bool) -> np.ndarray:
-        """Map a 2D array of probabilities to current IDs.
+    def align_probabilities(self, probabilities: np.ndarray | None) -> np.ndarray | None:
+        """Turn a cluster model's raw membership matrix into one column per current topic.
 
-        Arguments:
-            probabilities: 2D array of shape (n_samples, n_topics) to map.
-            from_original: If True, map from original IDs to current.
-                           If False, map from last applied mapping to current.
+        Cluster models emit a column per cluster in their own label order and none for
+        outliers, whose share is therefore whatever is left over. Columns are summed so
+        that clusters which have since been merged land on the same topic.
         """
-        if probabilities is not None:
-            return self.mapping.map_probabilities(probabilities, from_original=from_original)
-        else:
-            return None
+        if probabilities is None or probabilities.ndim != 2:
+            return probabilities
 
-    def get_mappings(self, from_original: bool = True) -> dict[int, int]:
-        """Get the current topic ID mappings.
+        topic_ids = self.topic_ids()
+        aligned = np.zeros((probabilities.shape[0], len(topic_ids)))
+        for label in range(probabilities.shape[1]):
+            topic_id = self.mapping.map(label)
+            if topic_id in topic_ids:
+                aligned[:, topic_ids.index(topic_id)] += probabilities[:, label]
 
-        Arguments:
-            from_original: If True, get mapping from original IDs to current.
-                           If False, get mapping from last applied mapping to current.
-        """
-        if from_original:
-            return self.mapping._mapping.copy()
-        else:
-            return self.mapping._recent_mapping.copy()
+        if -1 in topic_ids:
+            aligned[:, 0] = 1 - aligned.sum(axis=1)
+        return aligned
+
+    def get_mappings(self) -> dict[int, int]:
+        """Get the mapping from the cluster model's original labels to current topic IDs."""
+        return self.mapping._mapping.copy()
 
     def to_polars(self, topic: int | None = None) -> pl.DataFrame:
         """Convert topic info to a polars DataFrame."""
@@ -900,15 +860,12 @@ def to_dict(self, full: bool = False) -> dict:
             "bertopic_version": BERTOPIC_VERSION,
             "topics": {str(tid): topic.to_dict(full=full) for tid, topic in self.topics.items()},
             "mapping": self.mapping.to_dict(),
-            "predictions": self._original_predictions.tolist() if self._original_predictions.size > 0 else [],
+            "predictions": list(self.predictions),
             "actions": [a.value for a in self.actions],
         }
 
-        if full:
-            if self._original_probabilities is not None:
-                data["original_probabilities"] = self._original_probabilities.tolist()
-            if self._zeroshot_probabilities is not None:
-                data["zeroshot_probabilities"] = self._zeroshot_probabilities.tolist()
+        if full and self.probabilities is not None:
+            data["probabilities"] = self.probabilities.tolist()
 
         return data
 
@@ -918,14 +875,11 @@ def from_dict(cls, data: dict) -> "Topics":
         topics = cls()
         topics.topics = {int(tid): Topic.from_dict(td) for tid, td in data.get("topics", {}).items()}
         topics.mapping = TopicMapping.from_dict(data.get("mapping", {}))
-        topics._original_predictions = np.array(data.get("predictions", []))
+        topics.predictions = list(data.get("predictions", []))
         topics.actions = [TopicAction(a) for a in data.get("actions", [])]
 
-        # Handle full format fields
-        if "original_probabilities" in data:
-            topics._original_probabilities = np.array(data["original_probabilities"])
-        if "zeroshot_probabilities" in data:
-            topics._zeroshot_probabilities = np.array(data["zeroshot_probabilities"])
+        if "probabilities" in data:
+            topics.probabilities = np.array(data["probabilities"])
 
         return topics
 
@@ -999,8 +953,8 @@ def merge_similar(self, other: "Topics", min_similarity: float = 0.7) -> "Topics
         other_preds = [id_mapping[p] for p in other.predictions]
         all_preds = current_preds + other_preds
 
-        # Store as new "original" with identity mapping
-        self._original_predictions = np.array(all_preds)
+        # These rows are now the reference point, so the cluster mapping starts over
+        self.predictions = all_preds
         self.mapping.reset()
 
         # Recalculate document counts
@@ -1043,8 +997,8 @@ class TopicHierarchy:
         linkage_matrix: The scipy linkage matrix used to create the hierarchy.
         n_leaves: The number of leaf topics (excluding the outlier topic).
         outlier_topic: The outlier topic (if any). Not part of the hierarchy tree.
-        _original_predictions: Original document-to-leaf-topic predictions from fit.
-        _original_probabilities: Original document-to-leaf-topic probabilities from fit.
+        predictions: Row-to-leaf-topic assignments from the fit this hierarchy was built on.
+        probabilities: Row-to-leaf-topic probabilities from that same fit.
     """
 
     # Hierarchy structure
@@ -1056,8 +1010,8 @@ class TopicHierarchy:
     outlier_topic: Topic | None = None
 
     # Original values from fitting
-    _original_predictions: np.ndarray = field(default_factory=lambda: np.array([]))
-    _original_probabilities: np.ndarray | None = None
+    predictions: list[int] = field(default_factory=list)
+    probabilities: np.ndarray | None = None
 
     @property
     def root(self) -> Topic:
@@ -1136,12 +1090,11 @@ def _build_topics(self, selected_ids: list[int]) -> Topics:
             for leaf_id in topic.leaf_topic_ids:
                 old_to_new[leaf_id] = new_id
 
-        # Create Topics
+        # Create Topics, moving the rows onto the topic IDs at this level of the tree
         topics = Topics()
-        topics._original_predictions = self._original_predictions.copy()
-        topics._original_probabilities = (
-            self._original_probabilities.copy() if self._original_probabilities is not None else None
-        )
+        topics.predictions = list(self.predictions)
+        topics.probabilities = self.probabilities.copy() if self.probabilities is not None else None
+        topics._apply_to_rows(old_to_new, sorted(old_to_new))
         topics.mapping.apply(old_to_new)
 
         # Add outlier topic
@@ -1195,7 +1148,7 @@ def to_dict(self) -> dict:
             "linkage_matrix": self.linkage_matrix.tolist() if self.linkage_matrix.size > 0 else [],
             "n_leaves": self.n_leaves,
             "outlier_topic": self.outlier_topic.to_dict(full=True) if self.outlier_topic else None,
-            "predictions": self._original_predictions.tolist() if self._original_predictions.size > 0 else [],
+            "predictions": list(self.predictions),
         }
 
     @classmethod
@@ -1208,5 +1161,5 @@ def from_dict(cls, data: dict) -> "TopicHierarchy":
         hierarchy.outlier_topic = (
             Topic.from_dict(data["outlier_topic"]) if data.get("outlier_topic") else None
         )
-        hierarchy._original_predictions = np.array(data.get("predictions", []))
+        hierarchy.predictions = list(data.get("predictions", []))
         return hierarchy
diff --git a/bertopic/plotting/_distribution.py b/bertopic/plotting/_distribution.py
index 022b50c1..ad4482a2 100644
--- a/bertopic/plotting/_distribution.py
+++ b/bertopic/plotting/_distribution.py
@@ -58,9 +58,11 @@ def visualize_distribution(
             "probabilities that were supplied. Lower `min_probability` to prevent this error."
         )
 
-    # Get values and indices equal or exceed the minimum probability
-    labels_idx = np.argwhere(probabilities >= min_probability).flatten()
-    vals = probabilities[labels_idx].tolist()
+    # Get values and topics whose probability equals or exceeds the minimum. Column `j`
+    # holds topic `topic_ids()[j]`, which is not the same number once outliers exist.
+    topic_ids = topic_model._topics.topic_ids()
+    labels_idx = [topic_ids[column] for column in np.argwhere(probabilities >= min_probability).flatten()]
+    vals = probabilities[probabilities >= min_probability].tolist()
 
     # Create labels
     if isinstance(custom_labels, str):
@@ -70,7 +72,7 @@ def visualize_distribution(
         labels = ["_".join([label[0] for label in l[:4]]) for l in labels]  # noqa: E741
         labels = [label if len(label) < 30 else label[:27] + "..." for label in labels]
     elif topic_model.custom_labels_ is not None and custom_labels:
-        labels = [topic_model.custom_labels_[idx + topic_model._outliers] for idx in labels_idx]
+        labels = [topic_model.custom_labels_[topic_ids.index(topic)] for topic in labels_idx]
     else:
         labels = []
         for idx in labels_idx:
@@ -81,7 +83,7 @@ def visualize_distribution(
                 label = label[:40] + "..." if len(label) > 40 else label
                 labels.append(label)
             else:
-                vals.remove(probabilities[idx])
+                vals.remove(probabilities[topic_ids.index(idx)])
 
     # Create Figure
     fig = go.Figure(
diff --git a/bertopic/variations/_hierarchy.py b/bertopic/variations/_hierarchy.py
index 5bb537e7..2fe47c8b 100644
--- a/bertopic/variations/_hierarchy.py
+++ b/bertopic/variations/_hierarchy.py
@@ -132,8 +132,8 @@ def hierarchical_topics(
     hierarchy = TopicHierarchy(
         linkage_matrix=Z,
         n_leaves=n_leaves,
-        _original_predictions=np.array(topic_model.topics_),
-        _original_probabilities=topic_model.probabilities_,
+        predictions=list(topic_model.topics_),
+        probabilities=topic_model.probabilities_,
     )
 
     # Add outlier topic if it exists
diff --git a/bertopic/variations/_zeroshot.py b/bertopic/variations/_zeroshot.py
index bd941d87..2be5093f 100644
--- a/bertopic/variations/_zeroshot.py
+++ b/bertopic/variations/_zeroshot.py
@@ -124,7 +124,7 @@ def combine_zeroshot_topics(topic_model: "BERTopic", corpus: Corpus, zeroshot_da
 
         # Create new Topics
         topic_model._topics = Topics().initialize(corpus.topics, corpus._zeroshot_labels).sort_by_frequency()
-        corpus.map_topics_and_probabilities(topic_model._topics, from_original=True)
+        corpus.topics = topic_model._topics.predictions
         logger.info("Zeroshot Step 2 - Completed \u2713")
 
     return corpus
@@ -149,5 +149,5 @@ def update_probabilities(topic_model: "BERTopic", corpus: Corpus, zeroshot_docs:
         corpus.probabilities = (
             sim_matrix if topic_model.calculate_probabilities else np.max(sim_matrix, axis=1)
         )
-        topic_model._topics._zeroshot_probabilities = corpus.probabilities
+        topic_model._topics.probabilities = corpus.probabilities
     return corpus
diff --git a/tests/test_reduction/test_delete.py b/tests/test_reduction/test_delete.py
index bada34fe..2927eb86 100644
--- a/tests/test_reduction/test_delete.py
+++ b/tests/test_reduction/test_delete.py
@@ -22,9 +22,7 @@ def test_delete(model, request):
     # First deletion
     topics_to_delete = [1, 2]
     topic_model.delete_topics(topics_to_delete)
-    mappings = topic_model._topics.get_mappings(from_original=True)
-    original_predictions = topic_model._topics._original_predictions.tolist()
-    mapped_labels = [mappings[label] for label in original_predictions]
+    mappings = topic_model._topics.get_mappings()
 
     if model == "online_topic_model" or model == "kmeans_pca_topic_model":
         assert nr_topics == len(set(topic_model.topics_)) + 1
@@ -33,7 +31,7 @@ def test_delete(model, request):
         assert nr_topics == len(set(topic_model.topics_)) + 2
         assert sum(topic_model._topics.frequencies().values()) == length_documents
 
-    assert mapped_labels == topic_model.topics_
+    assert set(mappings.values()) <= set(topic_model._topics.topic_ids())
 
     # Find two existing topics for second deletion
     remaining_topics = sorted(list(set(topic_model.topics_)))
@@ -42,9 +40,7 @@ def test_delete(model, request):
 
     # Second deletion
     topic_model.delete_topics(topics_to_delete)
-    mappings = topic_model._topics.get_mappings(from_original=True)
-    original_predictions = topic_model._topics._original_predictions.tolist()
-    mapped_labels = [mappings[label] for label in original_predictions]
+    mappings = topic_model._topics.get_mappings()
 
     if model == "online_topic_model" or model == "kmeans_pca_topic_model":
         assert nr_topics == len(set(topic_model.topics_)) + 3
@@ -53,4 +49,4 @@ def test_delete(model, request):
         assert nr_topics == len(set(topic_model.topics_)) + 4
         assert sum(topic_model._topics.frequencies().values()) == length_documents
 
-    assert mapped_labels == topic_model.topics_
+    assert set(mappings.values()) <= set(topic_model._topics.topic_ids())
diff --git a/tests/test_reduction/test_merge.py b/tests/test_reduction/test_merge.py
index 4a641947..93d5d713 100644
--- a/tests/test_reduction/test_merge.py
+++ b/tests/test_reduction/test_merge.py
@@ -21,20 +21,16 @@ def test_merge(model, documents, request):
 
     topics_to_merge = [1, 2]
     topic_model.merge_topics(documents, topics_to_merge)
-    mappings = topic_model._topics.get_mappings(from_original=True)
-    original_predictions = topic_model._topics._original_predictions.tolist()
-    mapped_labels = [mappings[label] for label in original_predictions]
+    mappings = topic_model._topics.get_mappings()
 
     assert nr_topics == len(set(topic_model.topics_)) + 1
     assert sum(topic_model._topics.frequencies().values()) == len(documents)
-    assert mapped_labels == topic_model.topics_
+    assert set(mappings.values()) <= set(topic_model._topics.topic_ids())
 
     topics_to_merge = [1, 2]
     topic_model.merge_topics(documents, topics_to_merge)
-    mappings = topic_model._topics.get_mappings(from_original=True)
-    original_predictions = topic_model._topics._original_predictions.tolist()
-    mapped_labels = [mappings[label] for label in original_predictions]
+    mappings = topic_model._topics.get_mappings()
 
     assert nr_topics == len(set(topic_model.topics_)) + 2
     assert sum(topic_model._topics.frequencies().values()) == len(documents)
-    assert mapped_labels == topic_model.topics_
+    assert set(mappings.values()) <= set(topic_model._topics.topic_ids())
diff --git a/tests/test_topics.py b/tests/test_topics.py
index 3b1ef542..61d68a6f 100644
--- a/tests/test_topics.py
+++ b/tests/test_topics.py
@@ -55,7 +55,7 @@ def build_topics(counts: dict[int, int], probabilities: np.ndarray | None = None
     )
 
     if probabilities is not None:
-        topics._original_probabilities = probabilities
+        topics.probabilities = probabilities
 
     return topics
 
@@ -133,13 +133,9 @@ def test_sort_by_frequency_records_the_cumulative_mapping():
     topics = build_topics({-1: 5, 0: 2, 1: 8, 2: 4})
     topics.sort_by_frequency()
 
-    assert topics.get_mappings(from_original=True) == {-1: -1, 1: 0, 2: 1, 0: 2}
+    assert topics.get_mappings() == {-1: -1, 1: 0, 2: 1, 0: 2}
 
 
-@pytest.mark.xfail(
-    strict=True,
-    reason="TopicMapping.map_probabilities ignores the outlier column; fixed in unit 5",
-)
 def test_reordering_permutes_probability_columns():
     """Reordering topics permutes the columns without losing or duplicating mass."""
     probabilities = make_probabilities({-1: 0.1, 0: 0.2, 1: 0.6, 2: 0.1}, nr_documents=19)
@@ -150,10 +146,6 @@ def test_reordering_permutes_probability_columns():
     assert topics.probabilities[0].tolist() == pytest.approx([0.1, 0.6, 0.1, 0.2])
 
 
-@pytest.mark.xfail(
-    strict=True,
-    reason="TopicMapping.map_probabilities drops the outlier column; fixed in unit 5",
-)
 def test_reordering_preserves_total_probability_mass():
     """A permutation cannot change how much mass a document carries."""
     probabilities = make_probabilities({-1: 0.1, 0: 0.2, 1: 0.6, 2: 0.1}, nr_documents=19)
@@ -209,7 +201,7 @@ def test_zeroshot_topics_are_placed_before_clustered_topics():
 
     # Zero-shot topics keep their order even though topic 0 holds the fewest documents,
     # while the clustered topics 2, 3 and 4 are sorted by frequency behind them
-    assert topics.get_mappings(from_original=True) == {0: 0, 1: 1, 3: 2, 4: 3, 2: 4}
+    assert topics.get_mappings() == {0: 0, 1: 1, 3: 2, 4: 3, 2: 4}
     assert topics.labels[0] == "alpha"
     assert topics.labels[1] == "beta"
 
@@ -219,7 +211,7 @@ def test_zeroshot_only_model_keeps_every_topic_in_label_order():
     topics = Topics().initialize([0] * 1 + [1] * 7, zeroshot_labels=["alpha", "beta"])
     topics.sort_by_frequency()
 
-    assert topics.get_mappings(from_original=True) == {0: 0, 1: 1}
+    assert topics.get_mappings() == {0: 0, 1: 1}
     assert [topics[topic_id].topic_type for topic_id in topics.topic_ids()] == [
         TopicType.ZERO_SHOT,
         TopicType.ZERO_SHOT,
@@ -231,7 +223,7 @@ def test_clustered_only_model_is_sorted_purely_by_frequency():
     topics = Topics().initialize([0] * 1 + [1] * 7 + [2] * 3)
     topics.sort_by_frequency()
 
-    assert topics.get_mappings(from_original=True) == {1: 0, 2: 1, 0: 2}
+    assert topics.get_mappings() == {1: 0, 2: 1, 0: 2}
 
 
 def test_zeroshot_topics_survive_alongside_an_outlier():
@@ -242,7 +234,7 @@ def test_zeroshot_topics_survive_alongside_an_outlier():
 
     assert topics.topic_ids() == [-1, 0, 1, 2]
     assert topics[-1].topic_type == TopicType.OUTLIER
-    assert topics.get_mappings(from_original=True) == {-1: -1, 0: 0, 1: 1, 2: 2}
+    assert topics.get_mappings() == {-1: -1, 0: 0, 1: 1, 2: 2}
 
 
 # --------------------------------------------------------------------------------------
@@ -290,13 +282,9 @@ def test_merge_composes_with_an_earlier_reordering():
     topics.merge({-1: -1, 0: 0, 1: 0, 2: 1})
 
     # Original 1 and 2 were sorted to 0 and 1, then merged together into 0
-    assert topics.get_mappings(from_original=True) == {-1: -1, 1: 0, 2: 0, 0: 1}
+    assert topics.get_mappings() == {-1: -1, 1: 0, 2: 0, 0: 1}
 
 
-@pytest.mark.xfail(
-    strict=True,
-    reason="Corpus.map_probabilities sums but TopicMapping overwrites; fixed in unit 5",
-)
 def test_merge_sums_probability_columns():
     """Merging topics adds their probability mass together rather than discarding it."""
     probabilities = make_probabilities({-1: 0.1, 0: 0.2, 1: 0.6, 2: 0.1}, nr_documents=19)
@@ -361,10 +349,6 @@ def test_delete_accepts_a_single_topic_id():
     assert topics[-1].nr_documents == 9
 
 
-@pytest.mark.xfail(
-    strict=True,
-    reason="delete_topics never touches probabilities, leaving them stale; fixed in unit 5",
-)
 def test_delete_sums_probability_mass_into_the_outlier():
     """A deleted topic's probability mass moves to the outlier, mirroring its documents."""
     probabilities = make_probabilities({-1: 0.1, 0: 0.2, 1: 0.6, 2: 0.1}, nr_documents=19)
@@ -387,18 +371,9 @@ def test_mapping_composes_successive_operations():
     mapping.apply({0: 0, 1: 1, 2: 0})
 
     # Original 0 went to 2 then to 0, original 1 went to 0 then stayed, original 2 went to 1 then 1
-    assert mapping.map(0, from_original=True) == 0
-    assert mapping.map(1, from_original=True) == 0
-    assert mapping.map(2, from_original=True) == 1
-
-
-def test_mapping_reports_the_most_recent_step_separately():
-    """The recent mapping describes the last step only, which is what Corpus consumes."""
-    mapping = TopicMapping()
-    mapping.apply({0: 2, 1: 0, 2: 1})
-    mapping.apply({0: 0, 1: 1, 2: 0})
-
-    assert mapping.map(2, from_original=False) == 0
+    assert mapping.map(0) == 0
+    assert mapping.map(1) == 0
+    assert mapping.map(2) == 1
 
 
 def test_unknown_topic_ids_map_to_themselves():
@@ -406,7 +381,7 @@ def test_unknown_topic_ids_map_to_themselves():
     mapping = TopicMapping()
     mapping.apply({0: 1, 1: 0})
 
-    assert mapping.map(99, from_original=True) == 99
+    assert mapping.map(99) == 99
 
 
 def test_mapping_rejects_an_incomplete_new_mapping():
@@ -475,7 +450,7 @@ def test_round_trip_preserves_topics_and_mapping():
     assert restored.topic_ids() == topics.topic_ids()
     assert restored.frequencies() == topics.frequencies()
     assert restored.predictions == topics.predictions
-    assert restored.get_mappings(from_original=True) == topics.get_mappings(from_original=True)
+    assert restored.get_mappings() == topics.get_mappings()
     assert restored[0].representations["Main"].words == topics[0].representations["Main"].words
 
 
@@ -535,21 +510,26 @@ def test_hierarchy_round_trip_preserves_node_data():
 # --------------------------------------------------------------------------------------
 
 
-@pytest.mark.xfail(
-    strict=True,
-    reason="Corpus holds a second copy of assignments that needs manual syncing; fixed in unit 5",
-)
-def test_corpus_assignments_follow_topic_mutations_without_a_manual_sync():
-    """Reading assignments after a mutation must not require a separate sync step.
+def test_corpus_cannot_remap_assignments_itself():
+    """Assignments have one owner: `Topics` decides them, `Corpus` only carries them.
 
-    Today `Topics` and `Corpus` each hold document assignments, kept in step by hand
-    through `map_topics_and_probabilities` at nine call sites. Collapsing them to one
-    store is what makes this test pass.
+    The two used to be kept in step by hand at nine call sites, with probability
+    remapping implemented twice under contradicting semantics — one summing on merge,
+    the other overwriting. `Corpus` no longer has the machinery to remap anything, which
+    is what makes them unable to disagree.
     """
-    topics = build_topics({-1: 2, 0: 4, 1: 2})
-    corpus = Corpus(documents=[f"document {index}" for index in range(8)])
-    corpus.topics = np.array(topics.predictions)
+    assert not hasattr(Corpus, "map_topics")
+    assert not hasattr(Corpus, "map_probabilities")
+    assert not hasattr(Corpus, "map_topics_and_probabilities")
+
+
+def test_topics_keeps_rows_and_probabilities_consistent_through_a_merge():
+    """Every mutation moves the rows with the topics, in one step."""
+    probabilities = make_probabilities({-1: 0.1, 0: 0.2, 1: 0.6, 2: 0.1}, nr_documents=19)
+    topics = build_topics({-1: 5, 0: 8, 1: 4, 2: 2}, probabilities)
 
-    topics.merge({-1: -1, 0: 0, 1: 0})
+    topics.merge({-1: -1, 0: 0, 1: 0, 2: 1})
 
-    assert list(corpus.topics) == topics.predictions
+    assert set(topics.predictions) <= set(topics.topic_ids())
+    assert topics.probabilities.shape == (19, len(topics.topic_ids()))
+    assert topics.probabilities[0].sum() == pytest.approx(1.0)