Assure breakthrough time is always cast to ms

Reproduces and fixes #13962

Author: achaikou

src/ert/config/breakthrough_config.py

@@ -18,6 +18,15 @@ class BreakthroughConfig(DerivedResponseConfig):
     observed_dates: list[datetime] = Field(default_factory=list)
     has_finalized_keys: bool = True
 
+    @staticmethod
+    def response_schema() -> dict[str, Any]:
+        return {
+            "response_key": pl.String,
+            "threshold": pl.Float64,
+            "time": pl.Datetime(time_unit="ms", time_zone=None),
+            "values": pl.Float32,
+        }
+
     def derive_from_storage(
         self, iter_: int, realization: int, ensemble: Any
     ) -> pl.DataFrame:
@@ -55,7 +64,9 @@ def derive_from_storage(
             time_offset_series = pl.Series(breakthrough_time_offsets, dtype=pl.Float32)
         else:
             time_offset_series = pl.Series(breakthrough_time_offsets, dtype=Float32)
-            time_series = pl.Series(breakthrough_times).dt.cast_time_unit("ms")
+            time_series = pl.Series(breakthrough_times)
+
+        time_series = time_series.dt.cast_time_unit("ms")
 
         return pl.DataFrame(
             {
@@ -64,7 +75,7 @@ def derive_from_storage(
                 "time": time_series,
                 "values": time_offset_series,
             }
-        )
+        ).pipe(self._assert_schema, self.response_schema())
 
     @property
     def match_key(self) -> list[str]:

tests/ert/unit_tests/config/test_breakthrough_config.py

@@ -0,0 +1,66 @@
+from datetime import datetime
+
+import polars as pl
+
+from ert.config import BreakthroughConfig, SummaryConfig
+from ert.storage.local_storage import open_storage
+
+
+def test_that_derive_from_storage_frames_are_stackable_regardless_of_breakthrough_time(
+    tmp_path,
+):
+    with open_storage(tmp_path, mode="w") as storage:
+        response_key = "WWCT:OP1"
+        time = datetime(2000, 3, 2, 13, 0)  # ruff: ignore[call-datetime-without-tzinfo]
+
+        breakthrough_config = BreakthroughConfig(
+            keys=[f"BREAKTHROUGH:{response_key}"],
+            summary_keys=[response_key],
+            thresholds=[0.2],
+            observed_dates=[time],
+        )
+
+        summary_config = SummaryConfig(
+            keys=[response_key],
+            input_files=["not_relevant"],
+        )
+
+        experiment = storage.create_experiment(
+            experiment_config={
+                "response_configuration": [
+                    summary_config.model_dump(mode="json"),
+                    breakthrough_config.model_dump(mode="json"),
+                ],
+            }
+        )
+
+        ensemble = storage.create_ensemble(
+            experiment, ensemble_size=2, iteration=0, name="prior"
+        )
+
+        def create_summary_response_dataframe(
+            response_key: str, realization: int, value_modifier
+        ) -> pl.DataFrame:
+            return pl.DataFrame(
+                {
+                    "realization": [realization] * 5,
+                    "response_key": [response_key] * 5,
+                    "time": [datetime(2000, month, 1, 1, 0) for month in range(1, 6)],  # ruff: ignore[call-datetime-without-tzinfo]
+                    "values": [n / value_modifier for n in range(5)],
+                }
+            )
+
+        value_over_threshold = create_summary_response_dataframe(response_key, 0, 10)
+        value_under_threshold = create_summary_response_dataframe(response_key, 1, 100)
+
+        ensemble.save_response("summary", value_over_threshold, 0)
+        ensemble.save_response("summary", value_under_threshold, 1)
+
+        breakthrough_response0 = breakthrough_config.derive_from_storage(0, 0, ensemble)
+        breakthrough_response1 = breakthrough_config.derive_from_storage(0, 1, ensemble)
+
+        ensemble.save_response("breakthrough", breakthrough_response0, 0)
+        ensemble.save_response("breakthrough", breakthrough_response1, 1)
+
+        responses = ensemble.load_responses("breakthrough", (0, 1))
+        assert len(responses) == 2