test: shear-wave row validation, controlled-pass semantics, bounds

- shear_wave: validate_rows before screens (exact scheduled steps/t*, no
  duplicates, all numeric/list fields finite, nonnegative error fields,
  rho_min>0; signed amplitude/phase/interval errors keep their sign);
  INVALID_RUN for invalid refined rows; own_screens_pass vs controlled_pass
  so a CONTROL_SCREEN_FAIL set never reports a controlled pass; empty or
  duplicate refined schemes rejected; pressure-budget redundancy stated.
- run_case: SHEAR_WAVE_SHA256 repinned to the updated module.
- tests: divergence/advection bounds use a(t) and the derived h^2 term;
  invalid-row mutations; signed errors allowed; controlled_pass checks.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

tests/validation/taylor_green/shear_wave.py

@@ -23,8 +23,8 @@ INITIALIZATION = (
 # Per-row screens. phase_drift_abs, residual_l2 and the run-level rate screen
 # use eps(n) and are preregistered diagnostic choices, not error bounds.
 LIMITS = dict(velocity_l2=0.05, spurious_pressure_scaled=0.02, mass_relative=1e-5)
-# With equal absolute ceiling and floor (0.02) the comparative pressure budget
-# adds no constraint unless the root control exceeds 0.01; raw ratios are kept.
+# The comparative pressure budget max(2*control, 0.02) is never below the 0.02
+# absolute ceiling, so it is always redundant; raw control ratios are kept.
 FLOORS = dict(rate_error=0.002, residual_l2=0.002, spurious_pressure_scaled=0.02)
 BASELINE_FACTOR = 2.0
 
@@ -175,13 +175,51 @@ def fit_rate(rows, length, viscosity):
     )
 
 
+NONNEGATIVE = ("tstar", "mass", "mass_relative", "amplitude_exact", "modal_amplitude",
+               "phase_drift_abs", "velocity_l2", "residual_l2", "longitudinal_l2",
+               "spurious_pressure_rms", "spurious_pressure_scaled",
+               "spurious_pressure_max_scaled", "rho_min", "rho_max", "max_speed")
+SIGNED = ("a_hat", "b_hat", "amplitude_ratio", "amplitude_error", "phase_drift")
+
+
+def _number(value):
+    return (isinstance(value, (int, float)) and not isinstance(value, bool)
+            and math.isfinite(value))
+
+
+def validate_rows(record):
+    """Exact schedule, finite numbers (incl. lists) and nonnegative error fields.
+
+    Signed quantities (amplitudes, amplitude/phase errors) keep their sign.
+    Raises ValueError so callers can report an invalid run before any screen.
+    """
+    n, rows = record["n"], record["rows"]
+    steps = [round(tstar * n / U) for tstar in TIMES]
+    if not isinstance(rows, list) or len(rows) != len(steps):
+        raise ValueError("rows do not match the scheduled samples")
+    for row, step in zip(rows, steps):
+        if type(row.get("step")) is not int or row["step"] != step:
+            raise ValueError(f"unscheduled or duplicate step {row.get('step')!r}")
+        if not _number(row.get("tstar")) or abs(row["tstar"] - step * U / n) > 3e-17:
+            raise ValueError(f"wrong sample time at step {step}")
+        if not all(_number(row.get(key)) for key in (*NONNEGATIVE, *SIGNED)):
+            raise ValueError(f"missing or nonfinite diagnostic at step {step}")
+        if any(row[key] < 0 for key in NONNEGATIVE) or row["rho_min"] <= 0:
+            raise ValueError(f"negative error quantity or density at step {step}")
+        levels = row.get("spurious_pressure_scaled_by_level")
+        if (not isinstance(levels, list) or len(levels) != record["depth"]
+                or not all(_number(v) and v >= 0 for v in levels)):
+            raise ValueError(f"invalid per-level pressure at step {step}")
+
+
 def summarize(record):
+    validate_rows(record)
     rows = record["rows"]
     n = record["n"]
     out = fit_rate(rows, n, record["viscosity"])
     out.update(
         eps=eps(n),
-        complete=record["completed"] and len(rows) == len(TIMES),
+        complete=record["completed"] is True,
         residual_l2=max(r["residual_l2"] for r in rows),
         phase_drift_abs=max(r["phase_drift_abs"] for r in rows),
         spurious_pressure_scaled=max(r["spurious_pressure_scaled"] for r in rows),
@@ -190,7 +228,10 @@ def summarize(record):
         velocity_l2=max(r["velocity_l2"] for r in rows),
         mass_relative=max(r["mass_relative"] for r in rows),
     )
-    out["absolute_pass"] = out["complete"] and all(
+    if not all(_number(v) for v in (out["rate"], out["rate_error"], *out["interval_rate_errors"],
+                                     out["log_fit_residual_max"])):
+        raise ValueError("nonfinite rate fit")
+    out["own_screens_pass"] = out["complete"] and all(
         out[key] <= out["eps"] for key in ("rate_error", "phase_drift_abs", "residual_l2")
     ) and all(out[key] <= limit for key, limit in LIMITS.items())
     return out
@@ -199,36 +240,46 @@ def summarize(record):
 def _usable(record):
     try:
         out = summarize(record)
-    except (KeyError, ValueError):
+    except (KeyError, TypeError, ValueError):
         return None
-    finite = all(math.isfinite(v) for v in out.values() if isinstance(v, float))
-    return out if out["complete"] and finite else None
+    return out if out["complete"] else None
 
 
 def assess(base_control, fine_control, refined):
     """Uniform root-spacing and finest-spacing controls versus refined runs.
 
-    INVALID_CONTROL: missing, incomplete, nonfinite or mixed-source controls.
-    CONTROL_SCREEN_FAIL: complete finite controls that fail a screen; refined
-    results are still computed and kept.
+    INVALID_CONTROL: missing, incomplete, invalid-row or mixed-source controls.
+    INVALID_RUN: a refined record has invalid rows (schedule, NaN, sign).
+    CONTROL_SCREEN_FAIL: complete valid controls that fail their own screens;
+    refined own-screen results are kept, but no controlled comparison passes.
+    PASS/FAIL: every refined run's controlled_pass (own screens and budgets).
     """
-    if base_control is None or fine_control is None:
-        return dict(status="INVALID_CONTROL", reason="missing control")
+    if base_control is None or fine_control is None or not refined:
+        return dict(status="INVALID_CONTROL", reason="missing control or refined run")
     n = base_control["n"]
     if (base_control["depth"], fine_control["depth"], fine_control["n"]) != (1, 1, 2 * n):
         raise ValueError("controls must be uniform N and uniform 2N")
     if any(r["depth"] != 2 or r["n"] != n for r in refined):
         raise ValueError("refined runs must be depth 2 at root N")
+    if len({r.get("explosion") for r in refined}) != len(refined):
+        raise ValueError("duplicate refined explosion scheme")
     sources = {repr(r.get("source")) for r in [base_control, fine_control, *refined]}
     base, fine = _usable(base_control), _usable(fine_control)
     if base is None or fine is None or len(sources) != 1:
         return dict(status="INVALID_CONTROL",
-                    reason="incomplete/nonfinite control or mixed source")
-    report = dict(base_control=base, fine_control=fine, refined={}, status="PASS")
-    if not (base["absolute_pass"] and fine["absolute_pass"]):
-        report["status"] = "CONTROL_SCREEN_FAIL"
+                    reason="incomplete/invalid control or mixed source")
+    controls_pass = base["own_screens_pass"] and fine["own_screens_pass"]
+    report = dict(base_control=base, fine_control=fine, refined={},
+                  controls_pass=controls_pass)
+    statuses = []
     for r in refined:
-        s = summarize(r)
+        try:
+            s = summarize(r)
+        except (KeyError, TypeError, ValueError) as error:
+            report["refined"][r["explosion"]] = dict(valid=False, reason=str(error))
+            statuses.append("INVALID_RUN")
+            continue
+        s["valid"] = True
         s["budget"] = {
             key: max(BASELINE_FACTOR * base[key], floor) for key, floor in FLOORS.items()
         }
@@ -238,8 +289,13 @@ def assess(base_control, fine_control, refined):
                    for key in (*FLOORS, "spurious_pressure_rms")}
             for name, c in (("root", base), ("finest", fine))
         }
-        s["pass"] = s["absolute_pass"] and s["budget_pass"]
+        s["controlled_pass"] = controls_pass and s["own_screens_pass"] and s["budget_pass"]
         report["refined"][r["explosion"]] = s
-        if report["status"] == "PASS" and not s["pass"]:
-            report["status"] = "FAIL"
+        statuses.append("PASS" if s["controlled_pass"] else "FAIL")
+    if "INVALID_RUN" in statuses:
+        report["status"] = "INVALID_RUN"
+    elif not controls_pass:
+        report["status"] = "CONTROL_SCREEN_FAIL"
+    else:
+        report["status"] = "PASS" if all(v == "PASS" for v in statuses) else "FAIL"
     return report