Show local restart progress alongside full convergence threshold

experiments/plot_coupled_convergence.py

@@ -140,16 +140,26 @@ def load_pairs(paths, selected="both"):
         if name == "n2":
             require(bool(left.input_state_hashes) and left.input_state_hashes == right.input_state_hashes,
                     "Restart pair does not declare the same hash-bound input state files")
+            near(left.residual[0], right.residual[0], "shared restart initial norm")
         pairs[name] = (left, right)
     return pairs, sources
 
 
+def residual_reduction(candidate):
+    return [100. * (1. - value / candidate.residual[0]) for value in candidate.residual]
+
+
 def render_pair(name, curves):
     import matplotlib
     matplotlib.use("Agg")
     import matplotlib.pyplot as plt
 
-    figure, axis = plt.subplots(figsize=(8.5, 5.7), layout="constrained")
+    local = None
+    if name == "n2":
+        figure, (axis, local) = plt.subplots(2, 1, sharex=True, figsize=(8.5, 8.3),
+                                           gridspec_kw={"height_ratios": [1.35, 1.]}, layout="constrained")
+    else:
+        figure, axis = plt.subplots(figsize=(8.5, 5.7), layout="constrained")
     colors = ("#1769a8", "#c35b16")
     markers = ("o", "s")
     for candidate, color, marker in zip(curves, colors, markers):
@@ -164,6 +174,12 @@ def render_pair(name, curves):
         if failed:
             axis.scatter([candidate.time[i] for i in failed], [candidate.residual[i] for i in failed],
                          marker="x", s=65, linewidths=1.6, color=color, zorder=4)
+        if local is not None:
+            reduction = residual_reduction(candidate)
+            local.plot(candidate.time, reduction, color=color, marker=marker, markersize=4, linewidth=1.6)
+            if failed:
+                local.scatter([candidate.time[i] for i in failed], [reduction[i] for i in failed],
+                              marker="x", s=65, linewidths=1.6, color=color, zorder=4)
     if curves[0].threshold == curves[1].threshold:
         axis.axhline(curves[0].threshold, color="#555555", linestyle=":", linewidth=1.4,
                      label=f"Actual stopping threshold (both): {curves[0].threshold:.4g}")
@@ -173,7 +189,23 @@ def render_pair(name, curves):
                          label=f"{LABELS[candidate.identifier]} threshold: {candidate.threshold:.4g}")
     axis.set_yscale("log")
     axis.set_xlim(left=0., right=max(1., max(c.time[-1] for c in curves) * 1.04))
-    axis.set_xlabel("Cumulative reported Newton time (s)")
+    if local is None:
+        axis.set_xlabel("Cumulative reported Newton time (s)")
+    else:
+        # The full panel retains the stopping threshold. A separate zero-based
+        # local quantity reveals percent-level progress without compressing or
+        # hiding the many orders of magnitude still required for convergence.
+        all_reductions = [value for candidate in curves for value in residual_reduction(candidate)]
+        low, high = min(0., min(all_reductions)), max(0., max(all_reductions))
+        padding = .08 * max(high - low, .1)
+        local.set_ylim(low - padding if low < 0 else 0., high + padding)
+        local.set_ylabel("Residual reduction from\nshared restart (%)")
+        local.set_xlabel("Cumulative reported Newton time (s)")
+        local.set_title("Local progress from the same trajectories", fontsize=10)
+        local.grid(True, alpha=.22)
+        local.text(.02, .04, "Local reduction scale; actual stopping thresholds remain in the full-scale panel above.",
+                   fontsize=8, transform=local.transAxes,
+                   bbox={"facecolor": "white", "alpha": .8, "edgecolor": "none"})
     axis.set_ylabel("Normalized coupled residual")
     axis.grid(True, which="major", alpha=.22)
     axis.legend(fontsize=8, loc="best")
@@ -186,7 +218,7 @@ def render_pair(name, curves):
 
 
 def sidecar(name, curves, sources):
-    return {
+    report = {
         "pair": name, "source_audits": sources,
         "plotter_sha256": digest_bytes(Path(__file__).read_bytes()),
         "curves": [{"id": c.identifier, "campaign": c.campaign, "audit_path": c.audit_path,
@@ -202,6 +234,13 @@ def sidecar(name, curves, sources):
                   "Zero residuals or thresholds are refused rather than silently floored on the logarithmic axis.",
                   "The n=3 norm may include the central-density border; neither residual decline nor its stopping threshold proves physical equilibrium or virial balance."],
     }
+    if name == "n2":
+        report["local_progress_panel"] = {
+            "quantity": "100*(1-F/F0), in percent", "shared_initial_residual": curves[0].residual[0],
+            "zero_included": True, "actual_stopping_threshold_display": "full-scale logarithmic panel above",
+            "curves": [{"id": c.identifier, "residual_reduction_percent": residual_reduction(c)} for c in curves],
+        }
+    return report
 
 
 def main():

experiments/test_plot_coupled_convergence.py

@@ -109,6 +109,25 @@ class SchemaChecks(unittest.TestCase):
         with self.assertRaisesRegex(ValueError, "hash-bound"):
             plots.load_pairs([self.path])
 
+    def test_local_progress_starts_at_zero_and_retains_failed_endpoint(self):
+        pairs, sources = plots.load_pairs([self.path])
+        self.assertEqual(plots.residual_reduction(pairs["n2"][0]), [0., 50., 75., 75.])
+        local = plots.sidecar("n2", pairs["n2"], sources)["local_progress_panel"]
+        self.assertEqual(local["shared_initial_residual"], 1.)
+        self.assertTrue(local["zero_included"])
+        self.assertEqual(local["curves"][1]["residual_reduction_percent"], [0., 50., 75., 75.])
+
+    def test_shared_restart_must_also_have_matching_initial_norm(self):
+        candidate = self.data["campaigns"][0]["cases"][-1]
+        candidate["initial_residual"] *= 2
+        candidate["final_residual"] *= 2
+        candidate["actual_threshold"] *= 2
+        for row in candidate["history"]:
+            row["F"] = str(2 * float(row["F"]))
+        self.write()
+        with self.assertRaisesRegex(ValueError, "shared restart initial norm"):
+            plots.load_pairs([self.path])
+
     def test_nonpositive_log_values_are_not_silently_floored(self):
         self.data["campaigns"][0]["cases"][0]["history"][0]["F"] = "0"
         self.write()
@@ -135,6 +154,11 @@ class SchemaChecks(unittest.TestCase):
             self.assertEqual(axis.get_yscale(), "log")
             self.assertEqual(axis.lines[0].get_xdata()[0], 0.)
             self.assertEqual(list(axis.lines[-1].get_ydata()), [1e-8, 1e-8])
+            if name == "n2":
+                local = figure.axes[1]
+                self.assertEqual(local.get_yscale(), "linear")
+                self.assertEqual(local.get_ylim()[0], 0.)
+                self.assertEqual(list(local.lines[0].get_ydata()), [0., 50., 75., 75.])
             for extension in ("png", "pdf"):
                 path = self.directory / f"{name}.{extension}"
                 figure.savefig(path)