Audit actual stellar closure quadrature rules and support

experiments/audit_density_order.py

@@ -507,6 +507,26 @@ def physical(get, filename, meta):
     return result
 
 
+def audit_closure_rules(table, support):
+    rules = keyed(table, lambda row: int(row["element"]))
+    require(set(rules) == support, "Closure quadrature support differs from the stellar elements")
+    counts = defaultdict(int)
+    for row in rules.values():
+        require(set(row) == {"element", "quadrature_order", "quadrature_points"}, "Closure rule schema changed")
+        achieved, points = int(row["quadrature_order"]), int(row["quadrature_points"])
+        require(achieved > 0 and points > 0 and achieved % 2 == 1, "Invalid achieved tensor Gauss rule")
+        per_axis = (achieved + 1) // 2
+        require(points == per_axis ** 3, "Achieved order and tensor Gauss point count disagree")
+        counts[(achieved, points)] += 1
+    return {"scope": "Actual IntegrationRule achieved order and point count; the requested policy order is not exported.",
+            "stellar_elements": len(rules),
+            "distinct_achieved_orders": sorted({order for order, _ in counts}),
+            "distinct_point_counts": sorted({points for _, points in counts}),
+            "rules": [{"achieved_order": order, "quadrature_points": points,
+                       "points_per_axis": (order + 1) // 2, "elements": count}
+                      for (order, points), count in sorted(counts.items())]}
+
+
 def audit_case(campaign, name):
     get = lambda filename: campaign.artifact(name, filename)
     record = campaign.run(name)
@@ -532,6 +552,9 @@ def audit_case(campaign, name):
     audit_selfchecks(rows(get("density_order_root_selfchecks.csv")))
     controls = {(c, d) for c in CONTROLS for d in (2, 3, 4)} if frozen else {("stored_density", degree)}
     measures, support, totals, intervals = audit_measures(get, controls, n, K)
+    closure_rule_file = "production_closure_rules.csv" if frozen else "final_closure_rules.csv"
+    closure_rules = audit_closure_rules(rows(get(closure_rule_file)), support)
+    closure_rules["source_artifact"] = closure_rule_file
     error_quadrature = audit_error_quadrature(get, controls, support, totals, frozen)
     timing_rows = keyed(rows(get("density_order_timings.csv")), lambda r: r["phase"])
     require(set(timing_rows) == {"setup_and_projection", "independent_error_quadrature", "region_observation_and_output", "observer_total"},
@@ -557,6 +580,7 @@ def audit_case(campaign, name):
               "J": float(meta["J"]), "density_degree": degree, "state_size": len(state), "stellar_elements": len(support),
               "block_sizes": {f"{kind}/{block}": int(row["size"]) for (kind, block), row in blocks.items()},
               "measures": measures, "interval_audit": intervals, "unsliced_error_quadrature": error_quadrature,
+              "closure_quadrature": closure_rules,
               "observer_timings": timing,
               "run": {k: record[k] for k in ("exit_code", "seconds", "utc_elapsed_seconds", "utc_minus_monotonic_seconds", "child_cpu_seconds")}}
     if frozen:
@@ -631,6 +655,20 @@ def audit_case(campaign, name):
 
 
 class SelfTests(unittest.TestCase):
+    def test_actual_closure_rule_support_and_tensor_counts(self):
+        rows = [{"element": "2", "quadrature_order": "19", "quadrature_points": "1000"},
+                {"element": "5", "quadrature_order": "21", "quadrature_points": "1331"}]
+        result = audit_closure_rules(rows, {2, 5})
+        self.assertEqual(result["distinct_achieved_orders"], [19, 21])
+        self.assertEqual(result["distinct_point_counts"], [1000, 1331])
+        for altered in (rows[:1], rows + [rows[0]],
+                        [{**rows[0], "element": "3"}, rows[1]],
+                        [{**rows[0], "quadrature_order": "20"}, rows[1]],
+                        [{**rows[0], "quadrature_order": "0"}, rows[1]],
+                        [{**rows[0], "quadrature_points": "1001"}, rows[1]]):
+            with self.assertRaises(AssertionError):
+                audit_closure_rules(altered, {2, 5})
+
     def test_separate_enthalpy_observations(self):
         self.assertEqual(enthalpy_observations([], "h"),
                          {"samples": 0, "minimum_enthalpy": None, "negative_enthalpy_samples": 0})
@@ -747,6 +785,7 @@ def main():
                   "Complete negative partitions are checked only on lines with exported negative intervals; no global positivity claim.",
                   "Shallow unresolved witnesses remain recorded; root position tests have coefficient-scaled allowances.",
                   "Tangential convergence does not establish accuracy of the fixed eight-point normal rule; sliced first moments and inertia are compared observationally with q18.",
+                  "Density order alone changes among field spaces; production closure quadrature may also change under the unchanged order-aware policy. Only actual exported rules are reported.",
                   "Coupled timing and convergence gates are producer observations, not independent equation evaluations."],
               "campaigns": {}, "consumed_sha256": {str(Path(common.__file__).resolve()): digest(Path(common.__file__))}}
     comparable = defaultdict(list)