Classify temporal source series accurately

backend/app/services/source_freshness_service.py

@@ -101,6 +101,15 @@ def _is_local_storage_path(storage_path: str) -> bool:
     return bool(normalized) and "://" not in normalized and not normalized.startswith("/vsi")
 
 
+def _has_historical_series(datasets: list[Dataset]) -> bool:
+    observations_by_series: dict[str, set[datetime]] = defaultdict(set)
+    for dataset in datasets:
+        observed_at = _as_utc(dataset.observed_at)
+        if dataset.temporal_series_key and observed_at is not None:
+            observations_by_series[dataset.temporal_series_key].add(observed_at)
+    return any(len(observations) > 1 for observations in observations_by_series.values())
+
+
 def _integrity_summary(datasets: list[Dataset], versions_by_dataset: dict[UUID, list[DatasetVersion]]) -> SourceIntegritySummary:
     summary = SourceIntegritySummary()
     for dataset in datasets:
@@ -240,8 +249,6 @@ class SourceFreshnessService:
                 next_review_at = None
                 reason = "Voor deze bron is nog geen expliciete publicatie- of controlecyclus vastgelegd."
                 recommended_action = "Classificeer de bron eerst als snapshot, jaargang, vaste editie, scenario, archief of lokaal."
-            observed_values = {value for item in source_datasets if (value := _as_utc(item.observed_at)) is not None}
-            temporal_keys = {item.temporal_series_key for item in source_datasets if item.temporal_series_key}
             items.append(
                 SourceFreshnessItem(
                     source_name=source_name,
@@ -256,7 +263,7 @@ class SourceFreshnessService:
                     review_interval_days=policy.review_interval_days,
                     next_review_at=next_review_at,
                     status=status,
-                    historical_series=len(observed_values) > 1 or len(temporal_keys) > 1,
+                    historical_series=_has_historical_series(source_datasets),
                     reason=reason,
                     recommended_action=recommended_action,
                     integrity=integrity,

backend/tests/test_sprint221_source_freshness_audit.py

@@ -157,6 +157,30 @@ def test_rolling_orthophoto_prefers_explicit_current_snapshot_over_historical_ob
     assert report.items[0].latest_source_version == "most_recent_at_2026-07-14"
 
 
+def test_spatial_partitions_do_not_become_a_false_historical_series() -> None:
+    first = _dataset(
+        "dov_soil_map",
+        observed_at=datetime(2017, 6, 1, tzinfo=timezone.utc),
+        temporal_series_key="soil:mol",
+        source_version="2017",
+    )
+    second = _dataset(
+        "dov_soil_map",
+        observed_at=datetime(2017, 6, 1, tzinfo=timezone.utc),
+        temporal_series_key="soil:kempen",
+        source_version="2017",
+    )
+
+    report = SourceFreshnessService.build_report(
+        first.project_id,
+        [first, second],
+        [_version(first), _version(second)],
+        now=NOW,
+    )
+
+    assert report.items[0].historical_series is False
+
+
 def test_source_freshness_route_returns_canonical_envelope(monkeypatch) -> None:
     from app.api.routes import datasets as dataset_routes