test: preserve scalar receipts when optional shear probes fail

tests/validation/taylor_green/run_case.py

@@ -25,7 +25,7 @@ NESTED_SHEAR = "nested_shear_n128"  # fixed three-level/two-region shear preset
 SHEAR_NESTED_SHA256 = "fa6cfe87fb4df8c6d732c637c885276833c81973b950c7724d734f8fc350bcb0"
 TWIN_SHEAR = "twin_shear_n128"  # depth-2 twin: nested L0->L1 boxes without L2
 SHEAR_TWIN_SHA256 = "1cbf37f18e89aea577581631c4565f51e68d22a3739b698816ef64acbf2b2bb4"
-SHEAR_PROBES_SHA256 = "53ba8af8ac9c8078c40f84488c7ade4c11ccc91dd1fc275ff806042e9af30166"  # signed interface-probe mechanism diagnostic
+SHEAR_PROBES_SHA256 = "e2e6490d281cc275f448dc6d6ea7b9b4407cc74a51e9256b86c08c8626b89b76"  # signed interface-probe mechanism diagnostic
 
 
 def flow_functions(flow="taylor_green"):
@@ -382,9 +382,16 @@ def main():
                 run.step(step - run.step_count)
                 if step in probe_steps:
                     trace = report["probe_trace"]
-                    trace["steps"].append(probe_root[step])  # root-physical step
-                    trace["values"].append(shear_probes.sample(
-                        (run.download(),) if args.depth == 1 else run.download(), probes))
+                    if "error" not in trace:
+                        arrays = (run.download(),) if args.depth == 1 else run.download()
+                        try:
+                            values = shear_probes.sample(arrays, probes)
+                            shear_probes.validate_row(values, len(probes))
+                        except (ValueError, ZeroDivisionError) as error:
+                            report["probe_trace"] = dict(error=str(error))
+                        else:
+                            trace["steps"].append(probe_root[step])  # root-physical step
+                            trace["values"].append(values)
                     if step not in main_steps and step not in early_steps:
                         continue  # probe-only step: no qualification metrics
                 row = sample(run, masks, step, args.n, cfg.viscosity, initial_mass,
@@ -439,7 +446,7 @@ def main():
                         ),
                         flush=True,
                     )
-    if args.probe_trace:
+    if args.probe_trace and "error" not in report["probe_trace"]:
         try:
             report["probe_trace"]["bytes"] = shear_probes.validate(report["probe_trace"])
         except ValueError as error:  # keep the scalar receipt; drop an invalid trace

tests/validation/taylor_green/shear_probes.py

@@ -107,6 +107,12 @@ def residuals(trace):
     return out
 
 
+def validate_row(row, probe_count):
+    """Reject an invalid optional sample before it can enter a scalar receipt."""
+    if len(row) != 3 * probe_count or not all(math.isfinite(v) for v in row):
+        raise ValueError("malformed or nonfinite trace row")
+
+
 def validate(trace):
     """Exact schedule, finite values, fixed probe count and byte cap."""
     import json
@@ -114,10 +120,15 @@ def validate(trace):
         raise ValueError("trace steps differ from the fixed schedule")
     if len(trace["probes"]) != len(NORMAL_COORDS) * len(TRANSECTS):
         raise ValueError("unexpected probe count")
-    width = 3 * len(trace["probes"])
-    if any(len(r) != width or not all(math.isfinite(v) for v in r) for r in trace["values"]):
-        raise ValueError("malformed or nonfinite trace row")
-    size = len(json.dumps(trace, allow_nan=False).encode())
+    for row in trace["values"]:
+        validate_row(row, len(trace["probes"]))
+    # Include the final byte-count field itself, including its number of digits.
+    size = 0
+    while True:
+        final_size = len(json.dumps(dict(trace, bytes=size), allow_nan=False).encode())
+        if final_size == size:
+            break
+        size = final_size
     if size > TRACE_MAX_BYTES:
         raise ValueError(f"trace exceeds {TRACE_MAX_BYTES} bytes ({size})")
     return size

tests/validation/taylor_green/test_projected_pairs.py

@@ -429,7 +429,7 @@ class ProjectedPairsTests(unittest.TestCase):
                  patch.object(run_case, "runtime_identity", side_effect=AssertionError("CUDA reached")), \
                  redirect_stdout(stdout):
                 self.assertEqual(run_case.main(), 0)
-            verify.assert_called_once_with(REPO, MOCK_REVISION, "taylor_green")
+            verify.assert_called_once_with(REPO, MOCK_REVISION, "taylor_green", probes=False)
             self.assertTrue(json.loads(stdout.getvalue())["dry_run"])
             self.assertFalse(output.exists())
 
@@ -531,7 +531,7 @@ class ProjectedPairsTests(unittest.TestCase):
                     else:
                         self.assertEqual(run_case.main(), 0)
                 receipt = json.loads(output.read_text())
-                verify.assert_called_once_with(REPO, MOCK_REVISION, "taylor_green")
+                verify.assert_called_once_with(REPO, MOCK_REVISION, "taylor_green", probes=False)
                 self.assertEqual(selectors, [scheme] if depth == 2 else [])
                 self.assertEqual(receipt["source"], source())
                 self.assertEqual(receipt["runtime_identity"], RUNTIME)

tests/validation/taylor_green/test_shear_wave.py

@@ -1012,6 +1012,14 @@ class ProbeTraceTests(unittest.TestCase):
         trace["values"] = [[1.0, 0.01, -0.02] * 24 for _ in sp.STEPS]
         size = sp.validate(trace)
         self.assertLess(size, sp.TRACE_MAX_BYTES)
+        final_trace = dict(trace, bytes=size)
+        self.assertEqual(size, len(json.dumps(final_trace, allow_nan=False).encode()))
+        self.assertEqual(sp.validate(final_trace), size)
+        with patch.object(sp, "TRACE_MAX_BYTES", size):
+            self.assertEqual(sp.validate(trace), size)
+        with patch.object(sp, "TRACE_MAX_BYTES", size - 1):
+            with self.assertRaises(ValueError):
+                sp.validate(trace)
         realistic = dict(trace, values=[[1.0000012345678901, 0.012345678901234567,
                                          -0.023456789012345678] * 24 for _ in sp.STEPS])
         self.assertLess(sp.validate(realistic), sp.TRACE_MAX_BYTES)
@@ -1108,6 +1116,7 @@ class ProbeTraceTests(unittest.TestCase):
         self.assertEqual(trace["steps"], list(sp.STEPS))
         self.assertEqual(len(trace["values"]), len(sp.STEPS))
         self.assertLessEqual(trace["bytes"], sp.TRACE_MAX_BYTES)
+        self.assertEqual(trace["bytes"], len(json.dumps(trace, allow_nan=False).encode()))
         self.assertEqual(trace["case"], [run_case.NESTED_SHEAR, 128, 3, "projected_linear"])
         self.assertEqual(len(rec["rows"]), 5)  # qualification samples preserved
         self.assertEqual(files, ["probe.json"])  # no field dumps
@@ -1122,3 +1131,34 @@ class ProbeTraceTests(unittest.TestCase):
         self.assertTrue(all(p["level"] == 0 for p in t512["probes"]))
         res = sp.residuals(t512)
         self.assertLess(abs(res[0][0][1]), 1e-14)  # exact field at t=0 at actual centers
+
+    def test_invalid_optional_trace_preserves_scalar_rows(self):
+        sp = self.probes
+        case = (128, 1, run_case.NESTED_SHEAR, "constant", [np.ones((128, 128), bool)])
+        baseline, _, _ = self.run_mocked(*case)
+        original = sp.sample
+        for bad in ([float("nan")] * 72, [float("inf")] * 72, [1.0],
+                    ZeroDivisionError("zero probe density")):
+            calls = []
+
+            def sample(arrays, probes):
+                calls.append(True)
+                if len(calls) == 2:  # probe-only step 1, before the next scalar save
+                    if isinstance(bad, Exception):
+                        raise bad
+                    return bad
+                return original(arrays, probes)
+
+            with self.subTest(bad=repr(bad)), patch.object(sp, "sample", side_effect=sample):
+                rec, counts, files = self.run_mocked(*case)
+            self.assertTrue(rec["completed"])
+            self.assertEqual(rec["rows"], baseline["rows"])
+            self.assertEqual(set(rec["probe_trace"]), {"error"})
+            self.assertEqual(len(calls), 2)  # no further probe sampling after rejection
+            self.assertEqual(counts[-1], 2048)
+            self.assertEqual(files, ["probe.json"])
+        with patch.object(sp, "TRACE_MAX_BYTES", 100):
+            rec, _, _ = self.run_mocked(*case)
+        self.assertTrue(rec["completed"])
+        self.assertEqual(rec["rows"], baseline["rows"])
+        self.assertIn("exceeds", rec["probe_trace"]["error"])