Include the 2026 reporting year in the pipeline

src/igh_data_transform/transformations/candidates.py

@@ -26,6 +26,7 @@
     "new_includeinpipeline2021",
     "new_2023includeinevgendatabase",
     "new_2024includeinpipeline",
+    "new_includeinpipeline2025",
     "new_includeinpipeline",
 ]
 
@@ -172,7 +173,7 @@ def _resolve_rdstage_fk(
     # Strip product suffix: 'Phase III - Drugs' -> 'Phase III'
     lookup = lookup.str.rsplit(" - ", n=1).str[0]
     df = df.copy()
-    df["_resolved_rdstage_2025"] = df["_vin_currentrndstage_value"].map(lookup)
+    df["_resolved_rdstage_current"] = df["_vin_currentrndstage_value"].map(lookup)
     return df
 
 
@@ -214,20 +215,31 @@ def _expand_temporal_rows(df: pd.DataFrame) -> pd.DataFrame:
     valid_to = start of candidate's next boundary (None for latest).
     Must be called before column renaming (uses original bronze names).
     """
+    # R&D stage has no frozen 2025 archive column — unlike 2019/2021/2023/2024,
+    # IGH never froze one. `_resolved_rdstage_current` is Dataverse's rolling
+    # "current stage" field, so it sits at the 2025 boundary and the forward-fill
+    # in `_forward_fill` carries it into 2026. One rolling field therefore yields
+    # the same stage in both years, which is the honest reading of the source.
+    # If IGH ever freezes a 2025 stage column, add it here and move the rolling
+    # column to "2026-01-01".
     _rdstage_cols = [
         ("vin_2019stagepcr", "2019-01-01"),
         ("new_rdstage2021", "2021-01-01"),
         ("new_2023currentrdstage", "2023-01-01"),
         ("new_2024currentrdstage", "2024-01-01"),
-        ("_resolved_rdstage_2025", "2025-01-01"),
+        ("_resolved_rdstage_current", "2025-01-01"),
     ]
 
     _pipeline_cols = [
         ("vin_2019pcrpipelineinclusion", "2019-01-01"),
         ("new_includeinpipeline2021", "2021-01-01"),
         ("new_2023includeinevgendatabase", "2023-01-01"),
         ("new_2024includeinpipeline", "2024-01-01"),
-        ("new_includeinpipeline", "2025-01-01"),
+        # 2025 was frozen into its own column when IGH closed the collection.
+        ("new_includeinpipeline2025", "2025-01-01"),
+        # The unsuffixed column is Dataverse's rolling "current" field and now
+        # carries 2026. It rolls forward again each year.
+        ("new_includeinpipeline", "2026-01-01"),
     ]
 
     def _year_map(row: pd.Series, configs: list[tuple[str, str]]) -> dict:
@@ -249,7 +261,7 @@ def _forward_fill(year_map: dict, boundaries: list[str]) -> list:
         return result
 
     rows_out: list[dict] = []
-    cols_to_drop = _TEMPORAL_SOURCE_COLS + ["_resolved_rdstage_2025"]
+    cols_to_drop = _TEMPORAL_SOURCE_COLS + ["_resolved_rdstage_current"]
     # Columns to carry through (everything except temporal source cols)
     keep_cols = [c for c in df.columns if c not in cols_to_drop]
 
@@ -309,12 +321,12 @@ def transform_candidates(
     df = _normalize_pipeline_cols(df)
 
     # 1c. Capture the strict 2025 pipeline-inclusion value before temporal
-    # expansion consumes and drops `new_includeinpipeline`. The WHO
+    # expansion consumes and drops `new_includeinpipeline2025`. The WHO
     # Priority page needs the *actual* 2025 value (not the forward-filled
     # flag), so we preserve it candidate-grain under a stable name. Absent
     # column / NaN stays NaN; the gold CASE maps NaN and "No" to 0.
-    if "new_includeinpipeline" in df.columns:
-        df["includeinpipeline_2025_raw"] = df["new_includeinpipeline"]
+    if "new_includeinpipeline2025" in df.columns:
+        df["includeinpipeline_2025_raw"] = df["new_includeinpipeline2025"]
     else:
         df["includeinpipeline_2025_raw"] = None
 

src/igh_data_transform/transformations/silver_to_gold/config/schema_map.py

@@ -54,7 +54,7 @@
         "countries_approved_agg": "countries_product_approved",
         "candidate_type": "CASE WHEN captype_value = 'c1746ad3-93d1-f011-bbd3-00224892cefa' THEN 'Candidate' WHEN captype_value = '545d63d9-93d1-f011-bbd3-00224892cefa' THEN 'Product' ELSE 'Other' END",
         # Strict 2025 pipeline-inclusion flag for the WHO Priority page.
-        # 1 only when the raw 2025 `new_includeinpipeline` is exactly Yes;
+        # 1 only when the raw 2025 `new_includeinpipeline2025` is exactly Yes;
         # No, Pending, and blank all map to 0. Intentionally distinct from
         # the forward-filled `fact_pipeline_snapshot.include_in_pipeline`.
         "new_include_in_pipeline_2025": "CASE WHEN includeinpipeline_2025_raw = 100000000 THEN 1 ELSE 0 END",

tests/e2e/test_silver_to_gold_e2e.py

@@ -299,6 +299,51 @@ def test_pipeline_snapshot_candidate_key_not_all_null(self, gold_conn):
         df = _read_table(gold_conn, "fact_pipeline_snapshot")
         assert df["candidate_key"].notna().any(), "candidate_key is all NULL"
 
+    def test_fact_pipeline_snapshot_includes_2026(self, gold_conn):
+        """The 2026 reporting year reaches Gold. Before the rolling
+        pipeline-inclusion column was given its own boundary, 2026 values
+        were folded into 2025."""
+        years = pd.read_sql_query(
+            """
+            SELECT DISTINCT dt.year
+            FROM fact_pipeline_snapshot f
+            JOIN dim_date dt ON dt.date_key = f.date_key
+            WHERE dt.year IS NOT NULL
+            ORDER BY dt.year
+            """,
+            gold_conn,
+        )["year"].tolist()
+        assert 2026 in years, f"2026 missing from Gold; got {years}"
+
+    def test_2025_and_2026_read_different_source_columns(self, gold_conn):
+        """2025 must read the frozen `new_includeinpipeline2025` archive and
+        2026 the rolling `new_includeinpipeline`. Identical included-counts
+        would mean the rolling column is still feeding both boundaries —
+        the exact defect this change fixes.
+
+        Note this deliberately does NOT assert row-count parity between 2025
+        and 2024. All 244 candidates created during 2026 carry a rolling R&D
+        stage, and the stage series is pinned at 2025, so each contributes a
+        stage-only 2025 row. The raw `includeinpipeline` is NULL on those
+        rows, but the derivation maps NaN to 0, so those rows carry
+        `include_in_pipeline` 0 and are invisible to every portal query,
+        all of which filter on it.
+        """
+        counts = pd.read_sql_query(
+            """
+            SELECT dt.year, COUNT(*) AS n
+            FROM fact_pipeline_snapshot f
+            JOIN dim_date dt ON dt.date_key = f.date_key
+            WHERE dt.year IN (2025, 2026) AND f.include_in_pipeline = 1
+            GROUP BY dt.year
+            """,
+            gold_conn,
+        ).set_index("year")["n"]
+        assert counts.loc[2025] != counts.loc[2026], (
+            f"2025 and 2026 both report {counts.loc[2025]} included candidates; "
+            "the two boundaries appear to be reading the same source column"
+        )
+
     # -- fact_clinical_trial_event --
 
     def test_clinical_trial_has_rows(self, gold_conn):

tests/unit/test_bronze_to_silver.py

@@ -458,7 +458,7 @@ def test_lookup_tables_loaded_and_passed_for_candidates(self, tmp_path: Path):
         assert rows_2025["new_currentrdstage"].iloc[0] == "Phase III"
         # FK column should be dropped
         assert "_vin_currentrndstage_value" not in df.columns
-        assert "_resolved_rdstage_2025" not in df.columns
+        assert "_resolved_rdstage_current" not in df.columns
 
     def test_option_sets_loaded_and_passed_to_transformer(self, tmp_path: Path):
         """Option sets from Bronze are loaded and passed to transformers."""