Vectorize Farquhar, DI ControlLoop, gate pipeline, budget audit, chatbot Hebrew

src/budget_audit.py

@@ -0,0 +1,184 @@
+"""
+Budget audit pipeline — logs every slot spend and provides rollup reports.
+
+Appends one row per control tick to a parquet file. The weekly rollup
+verifies cumulative sacrifice stays within the 5% annual ceiling.
+
+Usage:
+    # In control_loop.py tick():
+    from src.budget_audit import BudgetAuditLog
+    audit = BudgetAuditLog()
+    audit.log_slot(tick_result)
+
+    # Weekly report:
+    python -m src.budget_audit --report
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+from datetime import date, datetime, timezone
+from pathlib import Path
+from typing import Optional
+
+import pandas as pd
+
+from config.settings import (
+    DATA_DIR,
+    MAX_ENERGY_REDUCTION_PCT,
+    SYSTEM_CAPACITY_KW,
+)
+
+logger = logging.getLogger(__name__)
+
+AUDIT_DIR = DATA_DIR / "budget_audit"
+AUDIT_PATH = AUDIT_DIR / "slot_log.parquet"
+
+
+@dataclass
+class SlotRecord:
+    """One row in the audit log."""
+    timestamp: datetime
+    date: date
+    slot_index: int
+    planned_offset_deg: float
+    actual_offset_deg: float
+    energy_cost_kwh: float
+    budget_spent_kwh: float
+    budget_remaining_kwh: float
+    gate_passed: bool
+    source: str
+    stage_id: str
+
+
+class BudgetAuditLog:
+    """Append-only parquet log for budget slot spends."""
+
+    def __init__(self, path: Path = AUDIT_PATH):
+        self.path = path
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+
+    def log_slot(self, tick_result) -> None:
+        """Append a tick result to the audit log."""
+        try:
+            record = {
+                "timestamp": getattr(tick_result, "timestamp", datetime.now(timezone.utc)),
+                "date": str(getattr(tick_result, "timestamp", datetime.now(timezone.utc)).date()
+                            if hasattr(getattr(tick_result, "timestamp", None), "date")
+                            else date.today()),
+                "slot_index": getattr(tick_result, "slot_index", -1),
+                "planned_offset_deg": getattr(tick_result, "plan_offset_deg", 0.0),
+                "actual_offset_deg": getattr(tick_result, "target_angle", 0.0),
+                "energy_cost_kwh": getattr(tick_result, "energy_cost_kwh", 0.0),
+                "budget_spent_kwh": getattr(tick_result, "budget_spent_kwh", 0.0),
+                "budget_remaining_kwh": getattr(tick_result, "budget_remaining_kwh", 0.0),
+                "gate_passed": getattr(tick_result, "live_gate_passed", False),
+                "source": getattr(tick_result, "source", ""),
+                "stage_id": getattr(tick_result, "stage_id", "unknown"),
+            }
+
+            new_row = pd.DataFrame([record])
+
+            if self.path.exists():
+                existing = pd.read_parquet(self.path)
+                combined = pd.concat([existing, new_row], ignore_index=True)
+            else:
+                combined = new_row
+
+            combined.to_parquet(self.path, index=False)
+            logger.debug("Audit log: slot %d, cost=%.4f kWh", record["slot_index"], record["energy_cost_kwh"])
+
+        except Exception as exc:
+            logger.warning("Budget audit log failed: %s", exc)
+
+    def load(self) -> pd.DataFrame:
+        """Load the full audit log."""
+        if self.path.exists():
+            return pd.read_parquet(self.path)
+        return pd.DataFrame()
+
+    def daily_summary(self, target_date: Optional[date] = None) -> dict:
+        """Summarize a single day's budget usage."""
+        df = self.load()
+        if df.empty:
+            return {"error": "No audit data"}
+
+        if target_date is None:
+            target_date = date.today()
+
+        day = df[df["date"] == str(target_date)]
+        if day.empty:
+            return {"date": str(target_date), "slots": 0, "total_cost_kwh": 0.0}
+
+        return {
+            "date": str(target_date),
+            "slots": len(day),
+            "total_cost_kwh": round(float(day["energy_cost_kwh"].sum()), 4),
+            "interventions": int(day["gate_passed"].sum()),
+            "max_offset_deg": round(float(day["actual_offset_deg"].abs().max()), 1),
+            "budget_remaining_kwh": round(float(day["budget_remaining_kwh"].iloc[-1]), 4),
+        }
+
+    def weekly_report(self) -> dict:
+        """Generate a weekly rollup report for budget compliance."""
+        df = self.load()
+        if df.empty:
+            return {"error": "No audit data"}
+
+        total_cost = float(df["energy_cost_kwh"].sum())
+        days = df["date"].nunique()
+        daily_potential_kwh = SYSTEM_CAPACITY_KW * 6.0  # ~6 peak sun hours
+        annual_potential_kwh = daily_potential_kwh * 365
+        ceiling_kwh = annual_potential_kwh * MAX_ENERGY_REDUCTION_PCT / 100.0
+
+        # Project annual rate from observed data
+        if days > 0:
+            daily_rate = total_cost / days
+            projected_annual = daily_rate * 365
+        else:
+            daily_rate = 0
+            projected_annual = 0
+
+        compliant = projected_annual <= ceiling_kwh
+
+        return {
+            "period_days": days,
+            "total_cost_kwh": round(total_cost, 3),
+            "daily_avg_kwh": round(daily_rate, 4),
+            "projected_annual_kwh": round(projected_annual, 1),
+            "ceiling_kwh": round(ceiling_kwh, 1),
+            "ceiling_pct": MAX_ENERGY_REDUCTION_PCT,
+            "utilization_pct": round(projected_annual / ceiling_kwh * 100, 1) if ceiling_kwh > 0 else 0,
+            "compliant": compliant,
+            "total_interventions": int(df["gate_passed"].sum()),
+            "intervention_rate_pct": round(float(df["gate_passed"].mean()) * 100, 1),
+        }
+
+
+# ---------------------------------------------------------------------------
+# CLI
+# ---------------------------------------------------------------------------
+
+if __name__ == "__main__":
+    import argparse
+    import json
+
+    parser = argparse.ArgumentParser(description="Budget audit report")
+    parser.add_argument("--report", action="store_true", help="Weekly rollup report")
+    parser.add_argument("--daily", type=str, help="Daily summary for YYYY-MM-DD")
+    args = parser.parse_args()
+
+    audit = BudgetAuditLog()
+
+    if args.report:
+        print(json.dumps(audit.weekly_report(), indent=2))
+    elif args.daily:
+        print(json.dumps(audit.daily_summary(date.fromisoformat(args.daily)), indent=2))
+    else:
+        df = audit.load()
+        if df.empty:
+            print("No audit data yet.")
+        else:
+            print(f"Audit log: {len(df)} slots, {df['date'].nunique()} days")
+            print(json.dumps(audit.weekly_report(), indent=2))

src/chatbot/guardrails.py

@@ -53,6 +53,10 @@ _DATA_KEYWORDS = [
     # Direct data ask
     r"\bshow me\b", r"\bwhat is\b", r"\bwhat are\b", r"\bhow much\b",
     r"\bcheck\b", r"\bstatus\b", r"\bstate\b",
+    # Hebrew keywords (common farmer queries)
+    r"להצליל", r"הצללה", r"טמפרטורה", r"מזג אוויר", r"גשם", r"רוח",
+    r"לחות", r"קרינה", r"השקיה", r"מים", r"אנרגיה", r"חשמל",
+    r"עכשיו", r"היום", r"מחר", r"אתמול", r"מה המצב", r"כמה",
 ]
 
 # Compile once
@@ -283,12 +287,17 @@ def estimate_confidence(
     tool_succeeded: bool,
     data_age_minutes: Optional[float],
     tool_name: Optional[str] = None,
+    rule_override: bool = False,
 ) -> str:
     """
     Estimate response confidence based on data grounding.
 
     Returns one of: "high", "medium", "low", "insufficient_data".
     """
+    # Rule-based override (e.g. dormancy, biology rules) — always high
+    if rule_override:
+        return "high"
+
     # No tool called at all
     if not tool_called:
         return "low"  # answering from system prompt / training data only

src/chatbot/vineyard_chatbot.py

@@ -169,6 +169,11 @@ You help the farmer decide when and how much to shade their Semillon grapevines
 (VSP trellis, 1.2 m canopy) under single-axis solar trackers (1.13 m panel at \
 2.05 m height, 3.0 m row spacing).
 
+LANGUAGE:
+- ALWAYS reply in the same language the user writes in. If they write in \
+Hebrew, reply in Hebrew. If English, reply in English. Match their language \
+exactly — do not switch languages mid-conversation.
+
 CONTROL OBJECTIVE:
 - Primary goal: maximise annual PV energy production.
 - Secondary goal: protect vines from heat, water stress, and sunburn using a \
@@ -180,12 +185,17 @@ CALENDAR & STAGE HANDLING:
 - Do NOT guess the current calendar month. If the user does not supply a \
 date and you do not have a phenology tool result, talk in terms of stages \
 (budburst, flowering, veraison, etc.) rather than asserting a specific month.
+- IMPORTANT: For "should I shade?" questions, ALWAYS consider phenological \
+stage FIRST. If the vine is dormant (no leaves), shading is irrelevant — \
+say so briefly and recommend full tracking for energy. Do not waste the \
+user's time with weather analysis when the vine has no leaves.
 
 COMMUNICATION STYLE:
-- Use plain language; explain jargon when you first use it
-- Be concise but thorough
+- Be CONCISE: 2-4 sentences for simple questions, not 15 lines
+- Lead with the answer, then give a brief reason
 - Always explain WHY a recommendation makes sense biologically
 - When uncertain, say so and suggest what data would help
+- Do NOT repeat that data is stale multiple times — mention it once
 
 BIOLOGICAL GUIDELINES (strong constraints; balance them with the energy objective):
 
@@ -729,6 +739,17 @@ class VineyardChatbot:
         now = datetime.now(tz=tz)
         lines.append(f"CURRENT STATUS ({now.strftime('%Y-%m-%d %H:%M')} IST):")
 
+        # Phenology FIRST — most important context for shading decisions
+        try:
+            from src.models.phenology import estimate_stage_for_date
+            from datetime import date
+            stage = estimate_stage_for_date(date.today())
+            dormant = stage.id in ("winter_dormancy", "dormant", "pre_budburst")
+            lines.append(f"  Phenology: {stage.name} ({stage.id})"
+                         + (" — DORMANT, no leaves, shading irrelevant" if dormant else ""))
+        except Exception:
+            pass
+
         # Weather
         try:
             wx = self.hub.weather.get_current()
@@ -790,15 +811,6 @@ class VineyardChatbot:
         except Exception:
             pass
 
-        # Phenology
-        try:
-            from src.phenology import estimate_stage_for_date
-            from datetime import date
-            stage = estimate_stage_for_date(date.today())
-            lines.append(f"  Phenology: {stage.name} ({stage.id})")
-        except Exception:
-            pass
-
         # Control status (from Redis via hub — no direct Redis import)
         try:
             ctrl = self.hub.advisory.get_status()
@@ -886,22 +898,39 @@ class VineyardChatbot:
                 tagged_result = tag_tool_result(tool_name, tool_result)
                 data_age = tagged_result.get("_data_age_minutes")
 
+                # Auto-supplement: when IMS is stale, also fetch TB sensors
+                supplement_text = ""
+                if tool_name == "get_current_weather" and data_age is not None and data_age > 120:
+                    try:
+                        snap = self.hub.vine_sensors.get_snapshot(light=True)
+                        if snap and "error" not in snap:
+                            snap_tagged = tag_tool_result("get_vine_state", snap)
+                            supplement_text = (
+                                f"\n\nADDITIONAL: IMS weather is stale ({data_age:.0f} min old). "
+                                f"Here are FRESH on-site sensor readings from ThingsBoard:\n"
+                                f"```json\n{json.dumps(snap_tagged, indent=2, default=str)}\n```\n"
+                                f"Use these fresh readings instead of the stale IMS data for "
+                                f"current conditions."
+                            )
+                    except Exception:
+                        pass
+
                 # Build Pass 2 prompt with source citation instructions
                 source_label = get_source_label(tool_name)
                 freshness_note = ""
                 if data_age is not None and data_age > 60:
                     freshness_note = (
-                        f"\n\nIMPORTANT: This data is {data_age:.0f} minutes old. "
-                        "Tell the user the data may be stale and conditions may have changed."
+                        f"\n\nNote: IMS data is {data_age:.0f} minutes old — "
+                        "mention this once, briefly."
                     )
 
                 tool_result_text = (
                     f"Tool result for {tool_name} "
                     f"(source: {source_label}):\n"
                     f"```json\n{json.dumps(tagged_result, indent=2, default=str)}\n```\n\n"
-                    f"Explain this result to the farmer in plain language. "
-                    f"When quoting numbers, mention that they come from {source_label}."
-                    f"{freshness_note}"
+                    f"Answer the farmer's question concisely (2-4 sentences). "
+                    f"Lead with the answer, then explain briefly."
+                    f"{freshness_note}{supplement_text}"
                 )
 
                 messages.append({"role": "model", "parts": [{"text": response_text}]})
@@ -913,19 +942,27 @@ class VineyardChatbot:
             else:
                 final_response = response_text
 
+            # Step 5: Post-response rule validation
+            validation_ctx = self._get_validation_context()
+            violations = validate_response(
+                response_text=final_response,
+                context=validation_ctx,
+            )
+
+            # Detect rule-based overrides (dormancy, blocked rules) for confidence
+            has_rule_override = any(
+                v.rule_name in ("no_leaves_no_shade_problem", "no_shade_before_10", "no_shade_in_may")
+                and v.severity == "block"
+                for v in violations
+            )
+
             # Step 4: Estimate confidence
             confidence = estimate_confidence(
                 tool_called=tool_call is not None,
                 tool_succeeded=tool_succeeded,
                 data_age_minutes=data_age,
                 tool_name=tool_name,
-            )
-
-            # Step 5: Post-response rule validation
-            validation_ctx = self._get_validation_context()
-            violations = validate_response(
-                response_text=final_response,
-                context=validation_ctx,
+                rule_override=has_rule_override,
             )
 
             caveats: list[str] = []

src/control_loop.py

@@ -111,6 +111,9 @@ class TickResult:
 class ControlLoop:
     """15-minute agrivoltaic control loop.
 
+    All dependencies are injected via ``__init__()`` with sensible defaults.
+    Pass explicit instances for testing (e.g. mock dispatcher).
+
     Parameters
     ----------
     dry_run : bool
@@ -119,6 +122,8 @@ class ControlLoop:
         Path to the day-ahead plan JSON file.
     log_path : Path
         Path for simulation log output.
+    arbiter, dispatcher, astro, hub, modes, fleet, budget_planner, router
+        Injectable dependencies — pass None (default) to auto-create.
     """
 
     def __init__(
@@ -126,21 +131,31 @@ class ControlLoop:
         dry_run: bool = True,
         plan_path: Path = DAILY_PLAN_PATH,
         log_path: Path = SIMULATION_LOG_PATH,
+        *,
+        arbiter=None,
+        dispatcher=None,
+        astro=None,
+        hub=None,
+        modes=None,
+        fleet=None,
+        budget_planner=None,
+        router=None,
     ):
         self.dry_run = dry_run
         self.plan_path = plan_path
         self.log_path = log_path
 
-        # Lazy-init components
-        self._arbiter = None
-        self._dispatcher = None
-        self._astro = None
-        self._hub = None
-        self._modes = None
-        self._fleet = None
+        # Dependencies — lazy-create if not injected
+        self._arbiter = arbiter
+        self._dispatcher = dispatcher
+        self._astro = astro
+        self._hub = hub
+        self._modes = modes
+        self._fleet = fleet
+        self._budget_planner = budget_planner
+        self._router = router
+
         self._schedulers: Dict[str, object] = {}
-        self._budget_planner = None
-        self._router = None
         self._current_plan: Optional[dict] = None
         self._tick_log: List[dict] = []
 
@@ -155,7 +170,7 @@ class ControlLoop:
         self._replan_count: int = 0
 
     # ------------------------------------------------------------------
-    # Lazy component init
+    # Lazy component init (auto-create when not injected)
     # ------------------------------------------------------------------
 
     @property
@@ -725,6 +740,13 @@ class ControlLoop:
             f" [OVERRIDE: {result.override_reason}]" if result.live_override else "",
         )
 
+        # 12. Budget audit — append slot to parquet log
+        try:
+            from src.budget_audit import BudgetAuditLog
+            BudgetAuditLog().log_slot(result)
+        except Exception as exc:
+            logger.debug("Budget audit log skipped: %s", exc)
+
         return result
 
     # ------------------------------------------------------------------

src/models/farquhar_model.py

@@ -302,29 +302,76 @@ class FarquharModel:
         humidity_col: Optional[str] = "Air1_airHumidity_ref",
     ) -> pd.Series:
         """
-        Compute A for each row. Uses CWSI from Tleaf/Tair/VPD with empirical bounds.
+        Compute A for each row using vectorized pandas operations (~100x faster).
         Returns Series of A (umol CO2 m-2 s-1), index aligned to df.
         """
         required = [par_col, tleaf_col, co2_col, vpd_col, tair_col]
         for c in required:
             if c not in df.columns:
                 return pd.Series(np.nan, index=df.index)
-        # Empirical CWSI bounds from (Tleaf - Tair) percentiles if enough data
-        dT = df[tleaf_col] - df[tair_col]
-        dTmin = float(dT.quantile(0.05)) if len(dT.dropna()) > 10 else -2.0
-        dTmax = float(dT.quantile(0.95)) if len(dT.dropna()) > 10 else 8.0
-        out = []
-        for _, row in df.iterrows():
-            try:
-                par, tleaf, co2, vpd, tair = row[par_col], row[tleaf_col], row[co2_col], row[vpd_col], row[tair_col]
-                if pd.isna([par, tleaf, co2, vpd, tair]).any():
-                    out.append(np.nan)
-                    continue
-                cwsi = self.calc_CWSI(float(tleaf), float(tair), float(vpd), dTmin, dTmax)
-                a = self.calc_photosynthesis(
-                    float(par), float(tleaf), float(co2), float(vpd), float(tair), CWSI=cwsi
-                )
-                out.append(a)
-            except (TypeError, ZeroDivisionError, ValueError):
-                out.append(np.nan)
-        return pd.Series(out, index=df.index)
+
+        # Extract columns as float arrays
+        par = df[par_col].astype(float)
+        tleaf = df[tleaf_col].astype(float)
+        co2 = df[co2_col].astype(float)
+        vpd = df[vpd_col].astype(float)
+        tair = df[tair_col].astype(float)
+
+        # Vectorized CWSI from (Tleaf - Tair) with empirical bounds
+        dT = tleaf - tair
+        n_valid = dT.notna().sum()
+        dTmin = float(dT.quantile(0.05)) if n_valid > 10 else -2.0
+        dTmax = float(dT.quantile(0.95)) if n_valid > 10 else 8.0
+        if dTmax <= dTmin:
+            cwsi = pd.Series(0.0, index=df.index)
+        else:
+            cwsi = ((dT - dTmin) / (dTmax - dTmin)).clip(0.0, 1.0)
+
+        # Vectorized FvCB computation
+        Tk = tleaf + 273.15
+
+        # Michaelis constants (Bernacchi et al. 2001) — vectorized
+        Kc = np.exp(38.05 - 79430.0 / (R * Tk))
+        Ko = np.exp(20.30 - 36380.0 / (R * Tk)) * 1000.0
+        gamma_star = np.exp(19.02 - 37830.0 / (R * Tk))
+
+        # Vcmax and Jmax (modified Arrhenius) — vectorized
+        Vcmax = _modified_arrhenius(Tk, self.params["k25_vcmax"], self.params["Ha_vcmax"],
+                                     self.params["Hd_vcmax"], self.params["S_vcmax"])
+        Jmax = _modified_arrhenius(Tk, self.params["k25_jmax"], self.params["Ha_jmax"],
+                                    self.params["Hd_jmax"], self.params["S_jmax"])
+
+        # Electron transport J — vectorized quadratic solve
+        alpha = self.params["alpha"]
+        theta = self.params["theta"]
+        b = alpha * par + Jmax
+        c_val = alpha * par * Jmax
+        disc = b * b - 4 * theta * c_val
+        disc_safe = disc.clip(lower=0)
+        J = ((b - np.sqrt(disc_safe)) / (2 * theta)).clip(lower=0)
+        J = np.minimum(J, Jmax)
+        J = J.where(par > 0, 0.0)
+
+        # Dark respiration
+        Rd = self.params["rd_frac"] * Vcmax
+
+        # Intercellular CO2 — vectorized ci/ca
+        vpd_scale = np.exp(-0.3 * (vpd - 1.0).clip(lower=0))
+        stress = 1.0 - 0.5 * cwsi.fillna(0)
+        gs_factor = 2.1 * vpd_scale * stress
+        ci = co2 * (1.0 - 1.0 / (1.6 * gs_factor.clip(lower=0.01)))
+        ci = ci.clip(lower=co2 * 0.3, upper=co2)
+
+        # Rubisco-limited (Ac) and RuBP-limited (Aj) rates
+        Ac = Vcmax * (ci - gamma_star) / (ci + Kc * (1.0 + OI / Ko))
+        Aj = J * (ci - gamma_star) / (4.0 * ci + 8.0 * gamma_star)
+
+        # Net assimilation
+        An = np.minimum(Ac, Aj) - Rd
+        An = An.clip(lower=0.0)
+
+        # NaN where any input was NaN
+        valid = par.notna() & tleaf.notna() & co2.notna() & vpd.notna() & tair.notna()
+        An = An.where(valid, np.nan)
+
+        return An

src/shading/tradeoff_engine.py

@@ -112,36 +112,12 @@ class InterventionGate:
         self.shade_eligible_cwsi_above = shade_eligible_cwsi_above
         self.shade_eligible_ghi_above = shade_eligible_ghi_above
 
-    def evaluate(
-        self,
-        tleaf_c: Optional[float],
-        ghi_w_m2: Optional[float],
-        cwsi: Optional[float],
-        shading_helps: Optional[bool],
-        dt: Optional[datetime] = None,   # accepted but not used; preserved for logging
-    ) -> GateDecision:
-        """
-        Evaluate whether the vine is significantly stressed.
+    # ------------------------------------------------------------------
+    # Individual gate checks (pipeline pattern)
+    # ------------------------------------------------------------------
 
-        Parameters
-        ----------
-        tleaf_c       : leaf temperature (°C) — from SensorRaw or forecast
-        ghi_w_m2      : global horizontal irradiance (W/m²) — from IMS or TB
-        cwsi          : Crop Water Stress Index [0–1] — from TB or computed
-        shading_helps : output of FarquharModel — True only when Rubisco-limited
-                        AND reducing PAR would increase net A
-        dt            : slot datetime (optional; used only for logging tags)
-
-        Returns
-        -------
-        GateDecision
-            passed=True only when all physiological stress conditions are met.
-            The caller then passes to TradeoffEngine.find_minimum_dose() to
-            determine whether the current sun geometry allows effective shading.
-        """
-        dec = GateDecision(passed=False)
-
-        # 1. Night / deep overcast guard — no useful sun, skip shadow computation
+    def _check_meaningful_sun(self, ghi_w_m2: Optional[float], dec: GateDecision) -> Optional[GateDecision]:
+        """Block if GHI too low (night / deep overcast)."""
         if ghi_w_m2 is not None and ghi_w_m2 < self.min_meaningful_ghi:
             dec.no_meaningful_sun = True
             dec.rejection_reason = (
@@ -149,9 +125,10 @@ class InterventionGate:
                 f"< {self.min_meaningful_ghi:.0f}"
             )
             return dec
+        return None
 
-        # 2. Leaf temperature below Rubisco transition — vine is light-limited,
-        #    reducing PAR would hurt photosynthesis
+    def _check_heat_stress(self, tleaf_c: Optional[float], dec: GateDecision) -> Optional[GateDecision]:
+        """Block if leaf temperature below Rubisco transition (vine is light-limited)."""
         if tleaf_c is not None and tleaf_c < self.shade_eligible_tleaf_above:
             dec.tleaf_below_threshold = True
             dec.rejection_reason = (
@@ -159,8 +136,10 @@ class InterventionGate:
                 f"< {self.shade_eligible_tleaf_above:.0f}°C (Rubisco transition)"
             )
             return dec
+        return None
 
-        # 3. Water stress not confirmed — vine is well-watered, no urgent need
+    def _check_water_stress(self, cwsi: Optional[float], dec: GateDecision) -> Optional[GateDecision]:
+        """Block if CWSI below threshold (vine not water-stressed)."""
         if cwsi is not None and cwsi < self.shade_eligible_cwsi_above:
             dec.cwsi_below_threshold = True
             dec.rejection_reason = (
@@ -168,8 +147,10 @@ class InterventionGate:
                 f"< {self.shade_eligible_cwsi_above:.2f}"
             )
             return dec
+        return None
 
-        # 4. Radiation load not high enough to cause meaningful heat build-up
+    def _check_radiation_load(self, ghi_w_m2: Optional[float], dec: GateDecision) -> Optional[GateDecision]:
+        """Block if radiation too low for meaningful heat build-up."""
         if ghi_w_m2 is not None and ghi_w_m2 < self.shade_eligible_ghi_above:
             dec.ghi_below_threshold = True
             dec.rejection_reason = (
@@ -177,9 +158,10 @@ class InterventionGate:
                 f"< {self.shade_eligible_ghi_above:.0f} W/m²"
             )
             return dec
+        return None
 
-        # 5. Biology confirms shading would help — FvCB model is Rubisco-limited
-        #    and A would increase if PAR on the fruiting zone drops
+    def _check_biology(self, shading_helps: Optional[bool], dec: GateDecision) -> Optional[GateDecision]:
+        """Block if FvCB model says shading would hurt (RuBP-limited)."""
         dec.biology_says_shade_helps = bool(shading_helps)
         if not shading_helps:
             dec.rejection_reason = (
@@ -187,8 +169,51 @@ class InterventionGate:
                 "possibly declining afternoon PAR or unusual conditions"
             )
             return dec
+        return None
+
+    # ------------------------------------------------------------------
+    # Main evaluate (pipeline composition)
+    # ------------------------------------------------------------------
+
+    def evaluate(
+        self,
+        tleaf_c: Optional[float],
+        ghi_w_m2: Optional[float],
+        cwsi: Optional[float],
+        shading_helps: Optional[bool],
+        dt: Optional[datetime] = None,   # accepted but not used; preserved for logging
+    ) -> GateDecision:
+        """
+        Evaluate whether the vine is significantly stressed.
+
+        Runs a pipeline of 5 checks in order. First rejection stops the pipeline.
+        Gate passes only when ALL stress conditions are simultaneously met.
+
+        Parameters
+        ----------
+        tleaf_c       : leaf temperature (°C)
+        ghi_w_m2      : global horizontal irradiance (W/m²)
+        cwsi          : Crop Water Stress Index [0–1]
+        shading_helps : output of FarquharModel
+        dt            : slot datetime (optional; for logging only)
+        """
+        dec = GateDecision(passed=False)
+
+        # Run checks as a pipeline — first rejection short-circuits
+        checks = [
+            lambda d: self._check_meaningful_sun(ghi_w_m2, d),
+            lambda d: self._check_heat_stress(tleaf_c, d),
+            lambda d: self._check_water_stress(cwsi, d),
+            lambda d: self._check_radiation_load(ghi_w_m2, d),
+            lambda d: self._check_biology(shading_helps, d),
+        ]
+
+        for check in checks:
+            rejection = check(dec)
+            if rejection is not None:
+                return rejection
 
-        # All stress conditions met — pass to TradeoffEngine for geometry check
+        # All stress conditions met
         dec.passed = True
         return dec