Add strict 2025 pipeline-inclusion flag to dim_candidate_core

src/igh_data_transform/transformations/candidates.py

@@ -307,12 +307,24 @@ def transform_candidates(
     # 1b. Normalize text-valued pipeline columns to integer codes
     df = _normalize_pipeline_cols(df)
 
+    # 1c. Capture the strict 2025 pipeline-inclusion value before temporal
+    # expansion consumes and drops `new_includeinpipeline`. 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"]
+    else:
+        df["includeinpipeline_2025_raw"] = None
+
     # 2. Temporal expansion (reads original bronze column names)
     df = _expand_temporal_rows(df)
 
     # 3. Drop columns (temporal sources already consumed by expansion)
     df = drop_columns_by_name(df, COLUMNS_TO_DROP)
-    df = drop_empty_columns(df, preserve=["valid_to", "valid_from"])
+    df = drop_empty_columns(
+        df, preserve=["valid_to", "valid_from", "includeinpipeline_2025_raw"]
+    )
 
     # 3b. Synthesize key clinical trial link
     if "new_ctregistrylink" in df.columns:

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

@@ -53,6 +53,11 @@
         "countries_approved_count": "numberofcountrieswithproductapproval",
         "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;
+        # 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",
         "indication": "indication",
         "indication_type": "OPTIONSET:indicationtype",
         "healthcare_facility_level": "OPTIONSET:healthcarefacilitylevel|vin_healthcarefacilitylevel",

src/igh_data_transform/transformations/silver_to_gold/core/ddl_generator.py

@@ -26,6 +26,10 @@
     "quarter",
     "enrollment_count",
     "option_code",
+    # Strict 2025 pipeline-inclusion flag: CASE WHEN evaluates to int 0/1,
+    # but the column name doesn't match any INTEGER_SUFFIXES pattern, so
+    # it must be listed here explicitly to get INTEGER affinity in SQLite.
+    "new_include_in_pipeline_2025",
 }
 
 

tests/e2e/test_silver_to_gold_e2e.py

@@ -176,6 +176,14 @@ def test_dim_candidate_core_test_format_is_label_not_code(self, gold_conn):
             f"{offenders.unique().tolist()[:5]}"
         )
 
+    def test_dim_candidate_core_strict_2025_flag(self, gold_conn):
+        """The strict 2025 flag exists and is strictly 0/1."""
+        df = _read_table(gold_conn, "dim_candidate_core")
+        assert "new_include_in_pipeline_2025" in df.columns
+        assert set(df["new_include_in_pipeline_2025"].dropna().unique()).issubset(
+            {0, 1}
+        )
+
     # -- dim_disease --
 
     def test_dim_disease_no_null_diseaseid(self, gold_conn):

tests/unit/test_candidates.py

@@ -1086,6 +1086,40 @@ def test_new_columns_not_dropped(self):
         # new_platform keeps its name (no prefix to strip)
         assert "new_platform" in result.columns
 
+    def test_includeinpipeline_2025_raw_preserved_at_candidate_grain(self):
+        """The raw 2025 inclusion value is captured into
+        `includeinpipeline_2025_raw` and copied onto every temporal row
+        for the candidate (not forward-filled, not dropped)."""
+        df = self._make_input_df()
+        lookup = self._make_lookup_tables()
+        result, _ = transform_candidates(df, lookup_tables=lookup)
+
+        assert "includeinpipeline_2025_raw" in result.columns
+
+        # Candidate A: bronze new_includeinpipeline == 100000000 (Yes)
+        rows_a = result[result["candidateid"] == "id-1"]
+        assert len(rows_a) > 1
+        assert (rows_a["includeinpipeline_2025_raw"] == 100000000.0).all()
+
+        # Candidate C: bronze new_includeinpipeline == 100000001 (No) —
+        # carried verbatim, NOT coerced.
+        rows_c = result[result["candidateid"] == "id-3"]
+        assert (rows_c["includeinpipeline_2025_raw"] == 100000001.0).all()
+
+    def test_includeinpipeline_2025_raw_survives_when_all_null(self):
+        """includeinpipeline_2025_raw is never dropped by drop_empty_columns
+        even when every candidate's 2025 value is null (all-None override)."""
+        df = self._make_input_df(
+            overrides={"new_includeinpipeline": [None, None, None]}
+        )
+        lookup = self._make_lookup_tables()
+        result, _ = transform_candidates(df, lookup_tables=lookup)
+
+        # Column must exist — if drop_empty_columns swallowed it, gold KeyErrors.
+        assert "includeinpipeline_2025_raw" in result.columns
+        # Every value should be null (the source was all-None).
+        assert result["includeinpipeline_2025_raw"].isna().all()
+
     def test_ctregistrylink_synthesis_applied(self):
         """CT registry link is cleaned during transform."""
         df = self._make_input_df(

tests/unit/test_silver_to_gold.py

@@ -90,3 +90,15 @@ def test_dedicated_to_women_or_children_uses_optionset_resolution(
             STAR_SCHEMA_MAP["dim_priority"]["dedicated_to_women_or_children"]
             == "OPTIONSET:crc8b_dedicatedtowomenorchildren"
         )
+
+    def test_new_include_in_pipeline_2025_is_strict_yes_only(self) -> None:
+        """`dim_candidate_core.new_include_in_pipeline_2025` is 1 only when
+        the raw 2025 value is exactly Yes (100000000); No/Pending/NULL → 0."""
+        from igh_data_transform.transformations.silver_to_gold.config.schema_map import (  # noqa: E501
+            STAR_SCHEMA_MAP,
+        )
+
+        assert (
+            STAR_SCHEMA_MAP["dim_candidate_core"]["new_include_in_pipeline_2025"]
+            == "CASE WHEN includeinpipeline_2025_raw = 100000000 THEN 1 ELSE 0 END"
+        )