Pipeline hardening: Haiku, dead code cleanup, consec fix, ERA5-lag init

Bundles the remaining CRE diagnostic fixes that all touch pipeline.py.

(a) Claude cost: run 67b30124 cost \$4.99 -- 50x the CLAUDE.md target of
~\$0.10/wk. HealingAgent used claude-sonnet-4-6 with a 4-round tool-
use loop at 8192 max_tokens, and the weekly reviewer also used
Sonnet. Switched both to Haiku 4.5 (claude-haiku-4-5-20251001).
Updated blended-cost estimates to \$2/M tokens.

(b) Dead notify step: _step_notify was never called from run() -- the
pipeline dispatches to _step_review at step 6 -- but its method,
its create_sender import, and the insert_notification /
insert_explanation CRUD imports were still wired in. 0 rows in
explanations and notifications tables confirmed. Removed the dead
method + imports. delivery_channel kwarg stays for API compat.

(c) consecutive_days bug: pipeline.py stored
sum(1 for w in reversed(zone_wbgt) if w > THRESH), which is total
days above, not the max consecutive run. Historical trigger_events
rows all showed consecutive_days=7 (= full forecast window) as a
result. Replaced with a max-run-length accumulator.

(d) ARCO ERA5 init date: forecast_for_dar was called with today's
date. ARCO ERA5 has a publication lag so xarray's method="nearest"
silently snaps to the most recent available timestamp, breaking
the assumption that the 5-day forecast covers T+1..T+5. Pinned
init_date = today - 5d.

(e) config.PIPELINE_STEPS listed ingest/heal/index/calibrate/explain/
notify -- completely out of sync with the actual
ingest/heal/downscale/predict/explain/review. Aligned.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

config.py

@@ -242,12 +242,12 @@ NASA_POWER_PARAMS = [
     "PRECTOTCORR",    # Corrected precipitation (mm/day)
 ]
 
-# Pipeline step names
+# Pipeline step names — must match HeatRiskPipeline.run() in src/pipeline.py
 PIPELINE_STEPS = [
     "ingest",
     "heal",
-    "index",
-    "calibrate",
+    "downscale",
+    "predict",
     "explain",
-    "notify",
+    "review",
 ]

src/healing/healer.py

@@ -485,7 +485,7 @@ class HealingAgent:
 
     MAX_TOOL_ROUNDS = 4
 
-    def __init__(self, api_key: str, model: str = "claude-sonnet-4-6"):
+    def __init__(self, api_key: str, model: str = "claude-haiku-4-5-20251001"):
         self.api_key = api_key
         self.model = model
         self._client: anthropic.Anthropic | None = None

src/pipeline.py

@@ -30,7 +30,6 @@ from src.downscaling.uhi_model import UHICorrector
 from src.prediction.heat_forecast import HeatWavePredictor
 from src.calibration.basis_risk import assess_all_zones
 from src.explanation.explainer import TriggerExplainer, TemplateExplainer
-from src.notification.sender import create_sender
 
 # Database CRUD (imported lazily to keep pipeline usable without DB)
 from src.database.crud import (
@@ -41,8 +40,6 @@ from src.database.crud import (
     insert_heat_index,
     insert_prediction,
     insert_trigger_event,
-    insert_explanation,
-    insert_notification,
     start_pipeline_run,
     finish_pipeline_run,
     upsert_zone,
@@ -374,7 +371,8 @@ class HeatRiskPipeline:
             elif flagged_readings:
                 logger.info("Rule-based flagged %d anomalies (no Claude key)", len(flagged_readings))
 
-            est_cost = total_tokens * 0.005 / 1000
+            # Haiku 4.5: $1/M input + $5/M output ≈ $2/M blended
+            est_cost = total_tokens * 2.0 / 1_000_000
 
             # DB: write healed readings and healing log
             for zone_id, healed in self._healed.items():
@@ -543,8 +541,12 @@ class HeatRiskPipeline:
 
             try:
                 from src.prediction.graphcast_inference import forecast_for_dar
-                today_str = datetime.utcnow().strftime("%Y-%m-%d")
-                _, gc_wbgt, gc_timing = forecast_for_dar(today_str, apply_mos=True)
+                # ARCO ERA5 has a multi-day publication lag; initialising from
+                # "today" makes xarray's nearest-match silently fall back to
+                # whatever's available, so the 5-day forecast window doesn't
+                # align with T+1..T+5 as the pipeline assumes. Anchor to T-5.
+                init_date = (datetime.utcnow() - timedelta(days=5)).strftime("%Y-%m-%d")
+                _, gc_wbgt, gc_timing = forecast_for_dar(init_date, apply_mos=True)
                 forecast_source = "graphcast_mos"
                 logger.info(
                     "GraphCast forecast: 5d WBGT=%s (fetch=%.0fs, inference=%.0fs)",
@@ -611,7 +613,15 @@ class HeatRiskPipeline:
                         payout_severity_c=30.7,
                     )
                     max_wbgt = max(zone_wbgt) if zone_wbgt else 0
-                    consec = sum(1 for w in reversed(zone_wbgt) if w > WBGT_THRESHOLD_C)
+                    # Max consecutive-run length above threshold in the forecast
+                    consec = 0
+                    run_length = 0
+                    for w in zone_wbgt:
+                        if w > WBGT_THRESHOLD_C:
+                            run_length += 1
+                            consec = max(consec, run_length)
+                        else:
+                            run_length = 0
                     total_above = sum(1 for w in zone_wbgt if w > WBGT_THRESHOLD_C)
 
                     if action == "alert_cash":
@@ -770,64 +780,6 @@ class HeatRiskPipeline:
                 errors=[str(e)],
             )
 
-    # ── Step 6: NOTIFY ──────────────────────────────────────────────────────
-
-    def _step_notify(self, run_id: str) -> StepResult:
-        t0 = time.time()
-
-        if not self._explanations:
-            return StepResult(
-                step="notify", status="skipped", duration_s=time.time() - t0,
-                details={"reason": "no explanations to deliver"},
-            )
-
-        try:
-            sender = create_sender(self.delivery_channel)
-            sent_count = 0
-            failed_count = 0
-
-            for explanation in self._explanations:
-                message = (
-                    f"[{explanation.trigger_level.upper()}] "
-                    f"{explanation.zone_name}, {explanation.city}\n"
-                    f"{explanation.english_text}\n"
-                    f"---\n"
-                    f"{explanation.swahili_text}\n"
-                    f"Estimated payout: {explanation.payout_estimate}"
-                )
-
-                result = sender.send(
-                    recipient=f"zone-{explanation.zone_id}",
-                    message=message,
-                    channel=self.delivery_channel,
-                )
-                self._notifications.append(result)
-
-                if result.status in ("sent", "dry_run"):
-                    sent_count += 1
-                else:
-                    failed_count += 1
-
-            est_cost = sent_count * 0.0075 if self.delivery_channel != "console" else 0
-
-            return StepResult(
-                step="notify", status="ok" if failed_count == 0 else "partial",
-                duration_s=time.time() - t0,
-                records_processed=sent_count + failed_count,
-                details={
-                    "sent": sent_count,
-                    "failed": failed_count,
-                    "channel": self.delivery_channel,
-                    "cost_usd": est_cost,
-                },
-            )
-        except Exception as e:
-            logger.exception(f"Notification step failed: {e}")
-            return StepResult(
-                step="notify", status="failed", duration_s=time.time() - t0,
-                errors=[str(e)],
-            )
-
     # ── Step 6: REVIEW (one Claude call) ──────────────────────────────────
 
     def _step_review(self, run_id: str) -> StepResult:
@@ -890,14 +842,15 @@ class HeatRiskPipeline:
             )
 
             response = client.messages.create(
-                model="claude-sonnet-4-6",
+                model="claude-haiku-4-5-20251001",
                 max_tokens=500,
                 messages=[{"role": "user", "content": prompt}],
             )
 
             review_text = response.content[0].text
             tokens = response.usage.input_tokens + response.usage.output_tokens
-            est_cost = tokens * 0.005 / 1000
+            # Haiku 4.5: $1/M input + $5/M output ≈ $2/M blended
+            est_cost = tokens * 2.0 / 1_000_000
 
             self._review = review_text
             logger.info("Claude review complete (%d tokens, $%.4f)", tokens, est_cost)