Phase 27: held-out-parity audit - penalty wiring + 3-layer failure isolation

Dress rehearsal (Docker main() over fake public+private with renamed device
IDs) exposed that one bad split could destroy the whole submission: a
check_dataset assertion killed the adaptive path AND the pickle fallback
identically, escaped main()'s handler, and no submission.csv was written.

- load_best_effort(): dataset consistency check demoted to a logged warning
- fallback guarded individually; new raw-file last resort (defer-everything
plans, alive-filtered) guarded again - a split can at worst lose itself
- cost_curves penalty constants (EARLY/LATE) wired from each scenario's
settings (evaluator charges these; constant 0.5/10.0 in train, byte-inert
at defaults); planning_window_days != 42 logs loudly
- audit: per-scenario settings vary only in base_location/base_room (16
bases; mobility sim already reads them live)

Rehearsal after fixes: both splits static-path, complete 2-split
submission.csv in 35s; all-48 checksum 1851.0186805555559 byte-exact.

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

EXPERIMENTS.md

@@ -986,6 +986,40 @@ Applies ONLY to the static held-out path; the panel champion (rolling,
 native fleet) keeps alpha=0.8, where the same sweep was decisively rejected
 (phases 1/6). The train path never reads ALPHA_STATIC - all-48 unaffected.
 
+## Phase 27 - held-out-parity audit & dress rehearsal (2 fixes shipped)
+
+Question: does our setup correctly reflect public/private mechanics? Audit of
+what can vary per scenario + a Docker dress rehearsal of the EXACT remote
+invocation (main() over fake public+private splits with renamed device IDs so
+the own-fleet guard routes to the static path).
+
+Findings:
+1. **Per-scenario settings DO vary in train** - 27 of 48 scenarios differ,
+   but only in base_location/base_room (16 distinct technician bases). The
+   mobility sim reads the live settings object, so this was already handled.
+   All penalties/limits/window are constant 0.5/10.0/42 across train.
+2. **Hardening shipped: penalty wiring.** cost_curves priced with hardcoded
+   EARLY/LATE while the evaluator charges per-scenario settings values - now
+   wired through per scenario (byte-inert at default values; checksum
+   confirms). planning_window_days != 42 now logs loudly (cannot be rewired:
+   the 42d grid is baked into features/quantiles).
+3. **Hardening shipped: failure isolation.** The rehearsal exposed that a
+   data-shaped exception (e.g. a check_dataset assertion) killed the adaptive
+   path AND the fallback identically - the exception escaped main(), and NO
+   submission.csv was written: one bad split destroyed all splits' plans.
+   Now three layers: load_best_effort (consistency check demoted to a logged
+   warning), individually guarded pickle fallback, and a raw-file last resort
+   (defer-everything plans; finite score, cannot poison other splits), itself
+   guarded so a split can at worst lose only itself.
+4. Rehearsal after fixes: public + private both take the static path (82
+   deaths, own-fleet 0.00), the injected inconsistency logs as a warning,
+   complete 2-split submission.csv in 35s. All-48 checksum byte-exact
+   (1851.0186805555559).
+5. Not verifiable locally, accepted as residual risk: held-out fleet size
+   (~4x per the metadata leak; count rule is a ratio, runtime proven by two
+   real submissions), scenario dates/weekdays, and the held-out label-lag
+   process (the ALPHA_STATIC=1.5 bet rides on the undercount direction).
+
 ## Test-look ledger
 
 8 looks total (now historical - see protocol change above); look 9 = the

batteryswap_example/submit.py

@@ -40,6 +40,7 @@ from batteryswap_example.features import (
     daily_voltage, daily_all, weekly_betas, master_curve, physics_features,
     features_at_origin, WINDOW_DAYS, BETA_DEFAULT)
 from batteryswap_example.models_cv import SklearnRUL, plan_tsroute
+from batteryswap_example import models_cv
 
 TRAIN_TABLE = Path(__file__).parent / 'data' / 'train_features.parquet'
 GRID_STEP_DAYS = 4
@@ -112,8 +113,23 @@ def build_split_table(locations, eol_times, scen_starts, daily, dall, betas,
     return table
 
 
+def load_best_effort(split_path, log):
+    """load_dataset with the consistency check demoted to a logged warning:
+    a benign set mismatch in a held-out split must not cost the whole split
+    (dress rehearsal: an assertion here killed adaptive AND the fallback,
+    which re-raises identically - one bad split -> empty submission)."""
+    from batteryswap_public.utils import check_dataset
+    locations, timeseries, eol_times, scenarios = load_dataset(
+        split_path, check=False)
+    try:
+        check_dataset(locations, timeseries, eol_times, scenarios)
+    except Exception as e:
+        log(f'  WARNING check_dataset: {e}')
+    return locations, timeseries, eol_times, scenarios
+
+
 def adaptive_split_plans(split, split_path, log=print, deadline=None):
-    locations, timeseries, eol_times, scenarios = load_dataset(split_path)
+    locations, timeseries, eol_times, scenarios = load_best_effort(split_path, log)
     scen_starts = [pd.Timestamp(s['start_time']) for s in scenarios]
     daily = daily_voltage(timeseries)
     dall = daily_all(timeseries)
@@ -174,6 +190,19 @@ def adaptive_split_plans(split, split_path, log=print, deadline=None):
             # runtime kill with a partial submission
             raise TimeoutError('split deadline exceeded mid-adaptive')
 
+        # cost_curves prices with module-level penalty constants; the
+        # evaluator charges whatever THIS scenario's settings say. Constant
+        # 0.5/10.0 across all train scenarios, but wire them through so a
+        # held-out scenario with different penalties is priced correctly.
+        stg = scenario['settings']
+        models_cv.EARLY = float(stg.early_replacement_penalty_daily)
+        models_cv.LATE = float(stg.late_replacement_penalty_daily)
+        if int(stg.planning_window_days) != WINDOW_DAYS:
+            # the 42-day grid is baked into features/labels/quantiles and
+            # cannot be rewired at runtime - flag it loudly and plan anyway
+            log(f'  WARNING {scenario["name"]}: planning_window_days='
+                f'{stg.planning_window_days} != {WINDOW_DAYS}')
+
         if static_model is not None:
             model, train_uniq = static_model, static_uniq
         elif T not in fitted:
@@ -238,7 +267,7 @@ def fallback_split_plans(split, split_path, log=print):
     log(f'  FALLBACK to static pickle for split={split}')
     with open(Path(__file__).parent / 'planners' / 'best.pickle', 'rb') as f:
         planner = pickle.load(f)
-    locations, timeseries, eol_times, scenarios = load_dataset(split_path)
+    locations, timeseries, eol_times, scenarios = load_best_effort(split_path, log)
     plans = []
     for scenario, locs, cut, eol in iterate_scenarios(
             locations, timeseries, eol_times, scenarios):
@@ -249,6 +278,32 @@ def fallback_split_plans(split, split_path, log=print):
     return plans
 
 
+def last_resort_split_plans(split, split_path, log=print):
+    """Defer-everything plans from raw files only (no load_dataset, no
+    model, no pickle): every battery alive at the scenario origin is placed
+    one day past the window, which the evaluator cuts - misses become
+    emergencies, but the split scores finitely and validly."""
+    import json
+    log(f'  LAST RESORT (defer plans) for split={split}')
+    devices = pd.read_csv(split_path / 'devices.csv')
+    eol = pd.read_csv(split_path / 'eol_times.csv')
+    eol['end_time'] = pd.to_datetime(eol['end_time'])
+    eol = eol.set_index('device_id')['end_time']
+    scenarios = json.load(open(split_path / 'scenarios.json'))
+    plans = []
+    for s in scenarios:
+        T = pd.Timestamp(s['start_time'])
+        e = eol.reindex(devices['device_id'])
+        alive = devices['device_id'][e.isna().values | (e > T).values]
+        plans.append(pd.DataFrame({
+            'day': (T + pd.Timedelta(days=WINDOW_DAYS + 1)).normalize(),
+            'battery': alive.values,
+            'split': split,
+            'scenario': s['name'],
+        }))
+    return plans
+
+
 def main():
     dataset_path = Path(os.environ.get('BATTERYSWAP_DATASET_PATH', '/tmp/data'))
     splits = os.environ.get('BATTERYSWAP_SPLITS', 'public,private').split(',')
@@ -270,7 +325,18 @@ def main():
                                                   log=log, deadline=deadline))
         except Exception:
             traceback.print_exc()
-            all_plans.extend(fallback_split_plans(split, dataset_path / split, log=log))
+            try:
+                all_plans.extend(
+                    fallback_split_plans(split, dataset_path / split, log=log))
+            except Exception:
+                traceback.print_exc()
+                try:
+                    all_plans.extend(
+                        last_resort_split_plans(split, dataset_path / split, log=log))
+                except Exception:
+                    # partial submission beats no submission: never let one
+                    # split's failure destroy the other splits' plans
+                    traceback.print_exc()
 
     submission = pd.concat(all_plans)[['day', 'battery', 'split', 'scenario']]
     submission.to_csv('submission.csv', index=False)