diff --git a/documentation/changelog.rst b/documentation/changelog.rst
index b9831a7c4f..d3ff854686 100644
--- a/documentation/changelog.rst
+++ b/documentation/changelog.rst
@@ -38,6 +38,7 @@ Infrastructure / Support
 Bugfixes
 -----------
 
+* Where several data sources report the same event, which one a search keeps is now decided the same way every time: a source version only counts against other versions of that source, and a caller that lists its sources gets the order it asked for [see `PR #2494 <https://www.github.com/FlexMeasures/flexmeasures/pull/2494>`_]
 * A KPI on the asset page counted an event once per data source that reported it, so a total could come out higher than any source reported; it now reduces one value per event, from the latest source version and the most recent belief within it [see `PR #2472 <https://www.github.com/FlexMeasures/flexmeasures/pull/2472>`_]
 * ``flexmeasures add schedule --dry-run`` no longer saves a schedule when it is combined with ``--as-job``, where the flag used to be dropped without a word and the queued job stored its schedule anyway; that combination is now rejected, and a dry run says how many beliefs it would have saved and which events they cover [see `PR #2483 <https://www.github.com/FlexMeasures/flexmeasures/pull/2483>`_]
 * Upgrading a database old enough to still carry the pre-``GenericAsset``/``Sensor`` tables now works, where the v0.18.0 migration that removes them crashed twice over: once while checking whether those tables hold data, as soon as one of them held more than a single row, and once while dropping them, because it dropped each table before the ones referencing it [see `PR #2475 <https://www.github.com/FlexMeasures/flexmeasures/pull/2475>`_]
diff --git a/flexmeasures/data/models/parsing_utils.py b/flexmeasures/data/models/parsing_utils.py
index df58f70f0f..68b6d0e09c 100644
--- a/flexmeasures/data/models/parsing_utils.py
+++ b/flexmeasures/data/models/parsing_utils.py
@@ -9,7 +9,7 @@
 from flexmeasures.data.models.data_sources import DataSource
 
 
-def parse_source_arg(
+def parse_source_arg_per_entry(
     source: (
         DataSource
         | int
@@ -19,38 +19,62 @@ def parse_source_arg(
         | Sequence[str]
         | None
     ),
-) -> list[DataSource] | None:
-    """Parse the "source" argument by looking up DataSources corresponding to any given ids or names.
+) -> list[list[DataSource]] | None:
+    """Parse the "source" argument, keeping the sources of each entry together.
+
+    One entry can name more than one source, because a name is not unique,
+    and the sources it names arrive in no particular order.
+    Callers that care which source is preferred should treat one entry as one preference,
+    rather than reading an order into what a single name happened to match.
 
     Passes None as is (i.e. no source argument is given).
-    Accepts ids and names as list or tuples, always converting them to a list.
     """
     if source is None:
         return source
     if isinstance(source, (DataSource, str, int)):
-        sources = [source]
+        entries: Sequence = [source]
     else:
-        sources = source
-    parsed_sources: list[DataSource] = []
-    for source in sources:
-        if isinstance(source, int):
-            parsed_source = db.session.get(DataSource, source)
+        entries = source
+    parsed_entries: list[list[DataSource]] = []
+    for entry in entries:
+        if isinstance(entry, int):
+            parsed_source = db.session.get(DataSource, entry)
             if parsed_source is None:
                 current_app.logger.warning(
-                    f"Beliefs searched for unknown source {source}"
+                    f"Beliefs searched for unknown source {entry}"
                 )
+                parsed_entries.append([])
             else:
-                parsed_sources.append(parsed_source)
-        elif isinstance(source, str):
-            _parsed_sources = db.session.scalars(
-                select(DataSource).filter_by(name=source)
-            ).all()
-            if _parsed_sources is []:
+                parsed_entries.append([parsed_source])
+        elif isinstance(entry, str):
+            named = db.session.scalars(select(DataSource).filter_by(name=entry)).all()
+            if not named:
                 current_app.logger.warning(
-                    f"Beliefs searched for unknown source {source}"
+                    f"Beliefs searched for unknown source {entry}"
                 )
-            else:
-                parsed_sources.extend(_parsed_sources)
+            parsed_entries.append(list(named))
         else:
-            parsed_sources.append(source)
-    return parsed_sources
+            parsed_entries.append([entry])
+    return parsed_entries
+
+
+def parse_source_arg(
+    source: (
+        DataSource
+        | int
+        | str
+        | Sequence[DataSource]
+        | Sequence[int]
+        | Sequence[str]
+        | None
+    ),
+) -> list[DataSource] | None:
+    """Parse the "source" argument by looking up DataSources corresponding to any given ids or names.
+
+    Passes None as is (i.e. no source argument is given).
+    Accepts ids and names as list or tuples, always converting them to a list.
+    """
+    parsed_entries = parse_source_arg_per_entry(source)
+    if parsed_entries is None:
+        return None
+    return [parsed_source for entry in parsed_entries for parsed_source in entry]
diff --git a/flexmeasures/data/models/reporting/tests/test_aggregator.py b/flexmeasures/data/models/reporting/tests/test_aggregator.py
index 044c9fa08b..e32f980fa2 100644
--- a/flexmeasures/data/models/reporting/tests/test_aggregator.py
+++ b/flexmeasures/data/models/reporting/tests/test_aggregator.py
@@ -189,6 +189,18 @@ def test_source_transition(setup_dummy_data, db):
     )  # the data from the first source is used
     assert (result[13:] == -1).all().event_value
 
+    # Naming the sources the other way round hands the overlapping event to the other source,
+    # which is what "the first source defined in the sources array" means.
+    reversed_result = agg_reporter.compute(
+        start=tz.localize(datetime(2023, 4, 24)),
+        end=tz.localize(datetime(2023, 4, 25)),
+        input=[dict(sensor=s3, sources=[ds2, ds1])],
+        output=[dict(sensor=report_sensor)],
+        belief_time=tz.localize(datetime(2023, 12, 1)),
+    )[0]["data"]
+    assert (reversed_result[:12] == 1).all().event_value
+    assert (reversed_result[12:] == -1).all().event_value
+
     # only considering DataSource 1
     result = agg_reporter.compute(
         start=tz.localize(datetime(2023, 4, 24)),
diff --git a/flexmeasures/data/models/time_series.py b/flexmeasures/data/models/time_series.py
index a22a3d9c62..fcfc28d45c 100644
--- a/flexmeasures/data/models/time_series.py
+++ b/flexmeasures/data/models/time_series.py
@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from collections.abc import Sequence
 from typing import Any, Type
 from datetime import datetime as datetime_type, timedelta
 from functools import cached_property
@@ -23,7 +24,10 @@
 from flexmeasures.data import db
 from flexmeasures.data.models.legacy_migration_utils import upgrade_value
 from flexmeasures.data.models.data_sources import keep_latest_version
-from flexmeasures.data.models.parsing_utils import parse_source_arg
+from flexmeasures.data.models.parsing_utils import (
+    parse_source_arg,
+    parse_source_arg_per_entry,
+)
 from flexmeasures.data.services.annotations import prepare_annotations_for_chart
 from flexmeasures.data.services.timerange import get_timerange
 from flexmeasures.data.queries.utils import get_source_criteria
@@ -302,7 +306,7 @@ def latest_state(
     ) -> tb.BeliefsDataFrame:
         """Search the most recent event for this sensor, and return the most recent ex-post belief.
 
-        :param source: search only beliefs by this source (pass the DataSource, or its name or id) or list of sources
+        :param source: search only beliefs by this source (pass the DataSource, or its name or id) or list of sources. Where a list is given, its order says which source to prefer for an event that several of them report.
         """
         return self.search_beliefs(
             horizons_at_most=timedelta(0),
@@ -431,7 +435,7 @@ def search_beliefs(  # noqa: C901
         :param beliefs_before: only return beliefs formed before this datetime (inclusive)
         :param horizons_at_least: only return beliefs with a belief horizon equal or greater than this timedelta (for example, use timedelta(0) to get ante knowledge time beliefs)
         :param horizons_at_most: only return beliefs with a belief horizon equal or less than this timedelta (for example, use timedelta(0) to get post knowledge time beliefs)
-        :param source: search only beliefs by this source (pass the DataSource, or its name or id) or list of sources. Without this set and a most recent parameter used (see below), the results can be of any source.
+        :param source: search only beliefs by this source (pass the DataSource, or its name or id) or list of sources. Without this set and a most recent parameter used (see below), the results can be of any source. Where a list is given, its order says which source to prefer for an event that several of them report, which matters when asking for one deterministic belief per event.
         :param user_source_ids: Optional list of user source ids to query only specific user sources
         :param source_account_ids: Optional account ID (or list thereof) to query only sources linked to specific accounts
         :param source_types: Optional list of source type names to query only specific source types *
@@ -440,7 +444,7 @@ def search_beliefs(  # noqa: C901
         :param most_recent_beliefs_only: only return the most recent beliefs for each event from each source (minimum belief horizon). Defaults to True.
         :param most_recent_events_only: only return (post knowledge time) beliefs for the most recent event (maximum event start). Defaults to False.
         :param most_recent_only: only return a single belief, the most recent from the most recent event. Fastest method if you only need one. Defaults to False. Setting this to True will turn off usage of most_recent_beliefs_only and most_recent_events_only. Use with care when data uses cumulative probability (more than one belief per event_start and horizon).
-        :param one_deterministic_belief_per_event: only return a single value per event (no probabilistic distribution and only 1 source)
+        :param one_deterministic_belief_per_event: only return a single value per event (no probabilistic distribution and only 1 source). Where several sources report an event, the one to keep is chosen by version within a family of sources sharing a name, type and model, and between families by the order the sources were passed in, then by the most recent belief, then by the highest source id.
         :param one_deterministic_belief_per_event_per_source: only return a single value per event per source (no probabilistic distribution)
         :param as_json: return beliefs in JSON format (e.g. for use in charts) rather than as BeliefsDataFrame
         :param compress_json: return beliefs, sensors and sources as separate datasets to be used for lookups
@@ -896,39 +900,118 @@ def to_epoch_list(column: str, np_dtype: str | None, divisor: int) -> list:
     return all_records, sources_metadata
 
 
+def _first_of_each_group(order: np.ndarray, keys: tuple[np.ndarray, ...]) -> np.ndarray:
+    """The positions, within `order`, of the first row of each group.
+
+    The rows are already sorted by `order`, so a group ends wherever one of its keys changes.
+    """
+    is_first = np.empty(len(order), dtype=bool)
+    is_first[:1] = True
+    if len(order) > 1:
+        changed = np.zeros(len(order) - 1, dtype=bool)
+        for key in keys:
+            sorted_key = key[order]
+            changed |= sorted_key[1:] != sorted_key[:-1]
+        is_first[1:] = changed
+    return order[is_first]
+
+
+def _belief_recency(bdf: tb.BeliefsDataFrame) -> np.ndarray:
+    """How recently each belief was formed, as a number that grows with recency."""
+    if "belief_time" in bdf.index.names:
+        return bdf.index.get_level_values("belief_time").asi8
+    # A shorter horizon means the belief was formed closer to the event, so later.
+    return -bdf.index.get_level_values("belief_horizon").asi8
+
+
 def _select_latest_version_and_belief_per_event(
     bdf: tb.BeliefsDataFrame,
+    preferred_sources: Sequence[DataSource | Sequence[DataSource]] | None = None,
 ) -> tb.BeliefsDataFrame:
-    """Keep, per event, the single belief with the latest source version,
-    breaking version ties by most recent belief time.
+    """Keep one belief per event, choosing between the sources that reported it.
+
+    The choice is made in two steps, because a version number orders the releases of one source,
+    and says nothing about a different source.
+    Sources are therefore grouped into families sharing a name, type and model,
+    and versions are only ever compared inside a family.
 
-    Beliefs that tie on both keep the order they came in,
-    which is what lets a caller express its own precedence by the order in which it passes its sources.
-    See `test_source_transition`, where the first source in the list wins the events both sources report.
+    Within a family, the latest version wins, then the most recent belief, then the highest source id.
 
-    Assumes deterministic beliefs (probabilistic depth 1) and a belief_time index level.
+    Between families, the order the caller named its sources in wins, for a caller that named any,
+    then the most recent belief, then the highest source id.
+    One entry of `preferred_sources` is one preference, so a group of sources given together,
+    as a name that matched several of them does, shares a rank rather than being ordered among itself.
+    That order is how a caller says which source it prefers where two of them report one event,
+    which is what `AggregatorReporter` offers through its `sources` field.
+
+    The highest source id is a last resort, so that a tie no one else broke is answered the same way every time,
+    and does not move when a source is renamed.
+
+    Assumes deterministic beliefs (probabilistic depth 1).
     """
+    if len(bdf) < 2:
+        return bdf
+
     source_codes, unique_sources = pd.factorize(bdf.index.get_level_values("source"))
-    versions = [
-        Version(source.version if source.version else "0.0.0")
-        for source in unique_sources
-    ]
-    version_ranks = {
+
+    families: dict = {}
+    family_per_source = np.array(
+        [
+            families.setdefault((source.name, source.type, source.model), len(families))
+            for source in unique_sources
+        ]
+    )
+    versions = [Version(source.version or "0.0.0") for source in unique_sources]
+    version_order = {
         version: rank for rank, version in enumerate(sorted(set(versions)))
     }
-    rank_per_row = np.array([version_ranks[version] for version in versions])[
-        source_codes
-    ]
-    event_values = bdf.index.get_level_values("event_start").asi8
-    belief_values = bdf.index.get_level_values("belief_time").asi8
-    # Sort by event (ascending), then version rank and belief time (both descending)
-    order = np.lexsort((-belief_values, -rank_per_row, event_values))
-    sorted_events = event_values[order]
-    is_first_of_event = np.empty(len(order), dtype=bool)
-    is_first_of_event[:1] = True
-    is_first_of_event[1:] = sorted_events[1:] != sorted_events[:-1]
-    mask = np.zeros(len(order), dtype=bool)
-    mask[order[is_first_of_event]] = True
+    version_per_source = np.array([version_order[version] for version in versions])
+    id_per_source = np.array(
+        [source.id if source.id is not None else -1 for source in unique_sources]
+    )
+    # Sources the caller did not name rank behind the ones it did, in the order it gave them.
+    # An entry that named nothing this database knows still holds its place,
+    # so the rank for the unnamed has to clear every entry, not merely the ones that matched.
+    entries = preferred_sources or []
+    positions: dict = {}
+    for position, entry in enumerate(entries):
+        # One source, or any sequence of them: test for the single case,
+        # so that the check cannot fall behind what the annotation allows.
+        group = [entry] if isinstance(entry, DataSource) else list(entry)
+        for source in group:
+            positions.setdefault(source.id, position)
+    unnamed_rank = len(entries)
+    position_per_source = np.array(
+        [positions.get(source.id, unnamed_rank) for source in unique_sources]
+    )
+
+    events = bdf.index.get_level_values("event_start").asi8
+    recency = _belief_recency(bdf)
+    family = family_per_source[source_codes]
+    version = version_per_source[source_codes]
+    source_id = id_per_source[source_codes]
+    position = position_per_source[source_codes]
+
+    # np.lexsort takes its keys least significant first.
+    within_family = _first_of_each_group(
+        np.lexsort((-source_id, -recency, -version, family, events)),
+        (events, family),
+    )
+    between_families = _first_of_each_group(
+        within_family[
+            np.lexsort(
+                (
+                    -source_id[within_family],
+                    -recency[within_family],
+                    position[within_family],
+                    events[within_family],
+                )
+            )
+        ],
+        (events,),
+    )
+    mask = np.zeros(len(bdf), dtype=bool)
+    mask[between_families] = True
     return bdf[mask]
 
 
@@ -1052,7 +1135,7 @@ def search(
         :param beliefs_before: only return beliefs formed before this datetime (inclusive)
         :param horizons_at_least: only return beliefs with a belief horizon equal or greater than this timedelta (for example, use timedelta(0) to get ante knowledge time beliefs)
         :param horizons_at_most: only return beliefs with a belief horizon equal or less than this timedelta (for example, use timedelta(0) to get post knowledge time beliefs)
-        :param source: search only beliefs by this source (pass the DataSource, or its name or id) or list of sources
+        :param source: search only beliefs by this source (pass the DataSource, or its name or id) or list of sources. Where a list is given, its order says which source to prefer for an event that several of them report.
         :param user_source_ids: Optional list of user source ids to query only specific user sources
         :param source_account_ids: Optional account ID (or list thereof) to query only sources linked to specific accounts
         :param source_types: Optional list of source type names to query only specific source types *
@@ -1061,7 +1144,7 @@ def search(
         :param most_recent_beliefs_only: only return the most recent beliefs for each event from each source (minimum belief horizon). Defaults to True.
         :param most_recent_events_only: only return (post knowledge time) beliefs for the most recent event (maximum event start)
         :param most_recent_only: only return a single belief, the most recent from the most recent event. Fastest method if you only need one.
-        :param one_deterministic_belief_per_event: only return a single value per event (no probabilistic distribution and only 1 source)
+        :param one_deterministic_belief_per_event: only return a single value per event (no probabilistic distribution and only 1 source). Where several sources report an event, the one to keep is chosen by version within a family of sources sharing a name, type and model, and between families by the order the sources were passed in, then by the most recent belief, then by the highest source id.
         :param one_deterministic_belief_per_event_per_source: only return a single value per event per source (no probabilistic distribution)
         :param resolution: Optional timedelta or pandas freqstr used to resample the results **
         :param sum_multiple: if True, sum over multiple sensors; otherwise, return a dictionary with sensors as key, each holding a BeliefsDataFrame as its value
@@ -1093,7 +1176,16 @@ def search(
             ).all()
             sensors.extend(sensors_from_names)
 
-        parsed_sources = parse_source_arg(source)
+        parsed_source_entries = parse_source_arg_per_entry(source)
+        parsed_sources = (
+            None
+            if parsed_source_entries is None
+            else [
+                parsed_source
+                for entry in parsed_source_entries
+                for parsed_source in entry
+            ]
+        )
         source_criteria = get_source_criteria(
             cls=cls,
             user_source_ids=user_source_ids,
@@ -1129,11 +1221,10 @@ def search(
                 custom_filter_criteria=source_criteria,
                 custom_join_targets=custom_join_targets,
             )
-            if use_latest_version_per_event:
-                bdf = keep_latest_version(
-                    bdf=bdf,
-                    one_deterministic_belief_per_event=one_deterministic_belief_per_event,
-                )
+            if use_latest_version_per_event and not one_deterministic_belief_per_event:
+                # Asking for one belief per event settles the versions itself, family by family,
+                # so it does not need this pass first.
+                bdf = keep_latest_version(bdf=bdf)
             if one_deterministic_belief_per_event:
                 if (
                     bdf.lineage.number_of_sources <= 1
@@ -1141,34 +1232,13 @@ def search(
                 ):
                     # Fast track, no need to loop over beliefs
                     pass
-                elif (
-                    bdf.lineage.probabilistic_depth == 1
-                    and "belief_time" in bdf.index.names
-                ):
-                    # Deterministic beliefs: no need to take the median,
-                    # just pick the winning belief per event directly
-                    bdf = _select_latest_version_and_belief_per_event(bdf)
                 else:
-                    # First make deterministic
-                    bdf = bdf.for_each_belief(get_median_belief)
-                    # Then sort each event by latest source version and most recent belief_time
-                    version_per_source = {
-                        source: Version(source.version if source.version else "0.0.0")
-                        for source in bdf.lineage.sources
-                    }
-                    bdf = bdf.sort_values(
-                        by=["event_start", "source", "belief_time"],
-                        ascending=[True, False, False],
-                        key=lambda col: (
-                            col.map(version_per_source) if col.name == "source" else col
-                        ),
+                    if bdf.lineage.probabilistic_depth != 1:
+                        # Make the beliefs deterministic, so that one of them can be chosen.
+                        bdf = bdf.for_each_belief(get_median_belief)
+                    bdf = _select_latest_version_and_belief_per_event(
+                        bdf, preferred_sources=parsed_source_entries
                     )
-                    # Finally, take the first belief for each event, thus preference latest version first, most recent belief_time second
-                    bdf = bdf[
-                        ~bdf.index.get_level_values("event_start").duplicated(
-                            keep="first"
-                        )
-                    ]
             elif one_deterministic_belief_per_event_per_source:
                 if len(bdf) == 0 or bdf.lineage.probabilistic_depth == 1:
                     # Fast track, no need to loop over beliefs
diff --git a/flexmeasures/data/tests/test_search_postprocessing.py b/flexmeasures/data/tests/test_search_postprocessing.py
index 8e794c6d5c..b0f88b277f 100644
--- a/flexmeasures/data/tests/test_search_postprocessing.py
+++ b/flexmeasures/data/tests/test_search_postprocessing.py
@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+from collections.abc import Sequence
 from datetime import timedelta
 
 import numpy as np
@@ -36,12 +37,44 @@ def make_random_deterministic_bdf(
 
 def naive_select_latest_version_and_belief_per_event(
     bdf: tb.BeliefsDataFrame,
+    preferred_sources: Sequence[DataSource | Sequence[DataSource]] | None = None,
 ) -> tb.BeliefsDataFrame:
-    """Reference implementation: per event, pick the belief with the latest
-    source version, breaking version ties by most recent belief time."""
-    winners: dict = {}
+    """Reference implementation, written per row rather than vectorised.
+
+    Per event, keep one belief per family of sources sharing a name, type and model,
+    choosing the latest version, then the most recent belief, then the highest source id.
+    Then, among those, choose the source the caller named first,
+    again falling back on the most recent belief and then the highest source id.
+    """
+    # One entry is one preference, and an entry that named nothing still holds its place,
+    # so the rank for the unnamed clears every entry rather than every source found.
+    entries = preferred_sources or []
+    positions: dict = {}
+    for position, entry in enumerate(entries):
+        # One source, or any sequence of them: test for the single case,
+        # so that the check cannot fall behind what the annotation allows.
+        group = [entry] if isinstance(entry, DataSource) else list(entry)
+        for source in group:
+            positions.setdefault(source.id, position)
+    unlisted = len(entries)
+
+    per_family: dict = {}
     for i, (event_start, belief_time, source, _cp) in enumerate(bdf.index):
-        candidate = (Version(source.version or "0.0.0"), belief_time)
+        family = (event_start, source.name, source.type, source.model)
+        candidate = (Version(source.version or "0.0.0"), belief_time, source.id or -1)
+        incumbent = per_family.get(family)
+        if incumbent is None or candidate > incumbent[0]:
+            per_family[family] = (candidate, i)
+
+    winners: dict = {}
+    for (event_start, _name, _type, _model), (_key, i) in per_family.items():
+        _, belief_time, source, _cp = bdf.index[i]
+        # A lower position is preferred, so it is negated to keep "greater is better".
+        candidate = (
+            -positions.get(source.id, unlisted),
+            belief_time,
+            source.id or -1,
+        )
         incumbent = winners.get(event_start)
         if incumbent is None or candidate > incumbent[0]:
             winners[event_start] = (candidate, i)
@@ -50,26 +83,182 @@ def naive_select_latest_version_and_belief_per_event(
 
 
 def test_select_latest_version_and_belief_per_event_equivalence():
+    """The vectorised choice agrees with the plainly written one, over random frames.
+
+    The sources span two families, so that the two steps of the choice are both exercised,
+    and each trial is run with and without a caller's preference.
+    """
     rng = np.random.default_rng(7)
     sources = [
+        DataSource(id=1, name="s1", model="model 1", type="forecaster", version=None),
         DataSource(
-            id=i + 1,
-            name="s1",
-            model="model 1",
-            type="forecaster",
-            version=version,
-        )
-        for i, version in enumerate([None, "0.1.0", "0.2.0", "0.2.0", "1.0.0"])
+            id=2, name="s1", model="model 1", type="forecaster", version="0.1.0"
+        ),
+        DataSource(
+            id=3, name="s1", model="model 1", type="forecaster", version="0.2.0"
+        ),
+        DataSource(
+            id=4, name="s1", model="model 1", type="forecaster", version="0.2.0"
+        ),
+        DataSource(id=5, name="s2", model="model 2", type="scheduler", version="1.0.0"),
+        DataSource(id=6, name="s2", model="model 2", type="scheduler", version="9.0.0"),
     ]
-    for trial in range(10):
-        bdf = make_random_deterministic_bdf(
-            rng, sources, n_beliefs=int(rng.integers(2, 30))
+    for preference in (
+        None,
+        [sources[4], sources[0]],
+        [sources[0], sources[5]],
+        # A name that matched two sources, given as one entry, ahead of a single source.
+        [[sources[1], sources[2]], sources[4]],
+        # An entry that matched nothing, before the ones that did.
+        [[], sources[5], sources[0]],
+    ):
+        for _ in range(10):
+            bdf = make_random_deterministic_bdf(
+                rng, sources, n_beliefs=int(rng.integers(2, 30))
+            )
+            result = _select_latest_version_and_belief_per_event(
+                bdf, preferred_sources=preference
+            )
+            expected = naive_select_latest_version_and_belief_per_event(
+                bdf, preferred_sources=preference
+            )
+            pd.testing.assert_frame_equal(pd.DataFrame(result), pd.DataFrame(expected))
+            # Exactly one belief per event
+            assert not result.index.get_level_values("event_start").duplicated().any()
+
+
+def _one_event_frame(
+    beliefs: list[tuple[DataSource, str, float]],
+) -> tb.BeliefsDataFrame:
+    """A frame of beliefs about one event, each given as its source, belief time and value."""
+    sensor = tb.Sensor("precedence sensor", event_resolution=timedelta(hours=1))
+    event_start = pd.Timestamp("2025-01-01T00:00:00+00:00")
+    return tb.BeliefsDataFrame(
+        [
+            tb.TimedBelief(
+                sensor=sensor,
+                source=source,
+                event_start=event_start,
+                belief_time=pd.Timestamp(belief_time),
+                event_value=value,
+            )
+            for source, belief_time, value in beliefs
+        ]
+    )
+
+
+def _chosen_value(beliefs, preferred_sources=None) -> float:
+    frame = _select_latest_version_and_belief_per_event(
+        _one_event_frame(beliefs), preferred_sources=preferred_sources
+    )
+    assert len(frame) == 1
+    return frame["event_value"].iloc[0]
+
+
+def test_a_later_version_of_one_source_wins_its_family():
+    """Within a family, the version says which code produced the value, so it comes first."""
+    old = DataSource(id=1, name="rep", model="Rep", type="reporter", version="1.0.0")
+    new = DataSource(id=2, name="rep", model="Rep", type="reporter", version="2.0.0")
+    # The older version spoke more recently, and still loses.
+    assert (
+        _chosen_value(
+            [(new, "2024-12-31T00:00+00:00", 2.0), (old, "2024-12-31T06:00+00:00", 1.0)]
         )
-        result = _select_latest_version_and_belief_per_event(bdf)
-        expected = naive_select_latest_version_and_belief_per_event(bdf)
-        pd.testing.assert_frame_equal(pd.DataFrame(result), pd.DataFrame(expected))
-        # Exactly one belief per event
-        assert not result.index.get_level_values("event_start").duplicated().any()
+        == 2.0
+    )
+
+
+def test_versions_are_not_compared_between_families():
+    """A version number orders one source's releases, and says nothing about another source.
+
+    A scheduler at v1 and a forecaster at v9 are unrelated numbering,
+    so the choice falls to the more recent belief instead.
+    """
+    scheduler = DataSource(
+        id=1, name="Seita", model="StorageScheduler", type="scheduler", version="1"
+    )
+    forecaster = DataSource(
+        id=2, name="Seita", model="Prophet", type="forecaster", version="9"
+    )
+    assert (
+        _chosen_value(
+            [
+                (scheduler, "2024-12-31T06:00+00:00", 1.0),
+                (forecaster, "2024-12-31T00:00+00:00", 9.0),
+            ]
+        )
+        == 1.0
+    )
+
+
+def test_a_caller_that_names_its_sources_says_which_it_prefers():
+    """Between families, the order the caller named its sources in decides, whatever the belief times say."""
+    meter = DataSource(id=1, name="meter", model="M", type="other")
+    scheduler = DataSource(id=2, name="Seita", model="S", type="scheduler")
+    beliefs = [
+        (meter, "2024-12-31T06:00+00:00", 1.0),
+        (scheduler, "2024-12-31T00:00+00:00", 2.0),
+    ]
+    assert _chosen_value(beliefs, preferred_sources=[scheduler, meter]) == 2.0
+    assert _chosen_value(beliefs, preferred_sources=[meter, scheduler]) == 1.0
+    # Naming neither leaves the more recent belief to decide.
+    assert _chosen_value(beliefs) == 1.0
+
+
+def test_sources_named_together_share_one_preference():
+    """A name can match several sources, and they arrive in no particular order.
+
+    Reading an order into that would let the database decide precedence,
+    so one entry is one preference, and the sources in it are told apart by belief time and id instead.
+    """
+    # Two sources of one name, which is what `source="rep"` would match.
+    older = DataSource(id=1, name="rep", model="A", type="reporter")
+    newer = DataSource(id=2, name="rep", model="B", type="reporter")
+    other = DataSource(id=3, name="other", model="C", type="reporter")
+    beliefs = [
+        (older, "2024-12-31T06:00+00:00", 1.0),
+        (newer, "2024-12-31T00:00+00:00", 2.0),
+        (other, "2024-12-31T12:00+00:00", 3.0),
+    ]
+    # Named first as one entry, the pair outranks the other source, and the fresher of the pair wins.
+    assert _chosen_value(beliefs, preferred_sources=[[older, newer], other]) == 1.0
+    # Whichever way round that entry lists them, since one entry is one preference.
+    assert _chosen_value(beliefs, preferred_sources=[[newer, older], other]) == 1.0
+    # Naming the other source first still puts it ahead of both.
+    assert _chosen_value(beliefs, preferred_sources=[other, [older, newer]]) == 3.0
+
+
+def test_an_entry_that_named_nothing_does_not_demote_the_ones_that_did():
+    """A caller can name a source this database does not know, and still be heard about the others.
+
+    An unknown id or name leaves an entry that matched nothing,
+    and that entry still holds its place, so the sources named after it must keep outranking the unnamed.
+    """
+    named = DataSource(id=1, name="scheduler", model="S", type="scheduler")
+    unnamed = DataSource(id=2, name="meter", model="M", type="other")
+    beliefs = [
+        (named, "2024-12-31T00:00+00:00", 1.0),
+        # The source nobody named holds the fresher belief, so only the naming can decide this.
+        (unnamed, "2024-12-31T06:00+00:00", 2.0),
+    ]
+    # Two entries matched nothing, and the third named the scheduler.
+    assert _chosen_value(beliefs, preferred_sources=[[], [], named]) == 1.0
+
+
+def test_a_tie_no_one_broke_is_answered_the_same_way_every_time():
+    """Two sources that nothing else tells apart are settled by the highest id.
+
+    The id does not move when a source is renamed, where sorting on the name would.
+    """
+    first = DataSource(id=1, name="zzz", model="M", type="reporter")
+    second = DataSource(id=2, name="aaa", model="M", type="reporter")
+    beliefs = [
+        (first, "2024-12-31T00:00+00:00", 1.0),
+        (second, "2024-12-31T00:00+00:00", 2.0),
+    ]
+    assert _chosen_value(beliefs) == 2.0
+    # And the same answer whichever order the rows arrive in.
+    assert _chosen_value(list(reversed(beliefs))) == 2.0
 
 
 def naive_compress_belief_records(df: pd.DataFrame, sensor_id: int):