Frank14f/CelerisLab
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test: opt-in scaled_shear_n256 four-case shear diagnostic

Browse Source 
- run_case: --diagnostic scaled_shear_n256 (requires --flow shear_wave;
  U256/U512/constant R256/projected R256, Re100, no early trace); N512
  only via one of the two scaled diagnostics; scaled_n256 stays TG-only.
- assess_shear_wave: scaled=True contract reusing 92b8afe scaled
  case/build/workload checks, shear diagnostic identity; signed rate bias,
  refined excess over each control; hash-pinned de5b0cc N128 reference and
  descriptive N128->N256 signed trend; CLI --diagnostic/--reference-files.
- tests: scaled contract, set mixing, trend/reference pin, runner options.
- shear_wave.py and oracle.py unchanged.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Claude
2026-10-19 01:14:54 +00:00
co-authored by Claude Opus 5.5
parent 46100e1d53
commit 5fb95f5f4d
4 changed files with 263 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/validation/taylor_green.md
+42
View File
@@ -295,3 +295,45 @@ is not placement robustness. Two resolutions show a trend, not an order.
- Merge note: native 92b8afe adds a `scaled_n256` receipt block. When the
two are merged, keep both blocks and make `--flow shear_wave` and
`scaled_n256` mutually exclusive. Not merged here.
### Scaled shear diagnostic (opt-in `scaled_shear_n256`)
Runner: `run_case.py --flow shear_wave --diagnostic scaled_shear_n256 --re 100`
with one of these four cases:
- `--n 256 --depth 1` (U256)
- `--n 512 --depth 1` (U512)
- `--n 256 --depth 2 --explosion constant` (constant R256)
- `--n 256 --depth 2 --explosion projected_linear` (projected R256)
Everything physical is unchanged from N128:
- `rectangles(256,2)` covers the same physical box;
- same t* schedule (16N root steps);
- `nu = U*N/100`;
- same `eps(N)`, limits and floors.
`scaled_n256` (Taylor–Green, from native 92b8afe) and `scaled_shear_n256`
each require their own flow, so they are mutually exclusive. N512 is
accepted only through one of the two diagnostics. The shear diagnostic
takes no early-pressure trace.
Assessment:
python assess_shear_wave.py U256.json U512.json R256_constant.json R256_projected.json \
--diagnostic scaled_shear_n256 --reference-files <four de5b0cc N128 receipts> \
--output report.json --expected-revision <sha>
- The scaled N256/512 case, build and workload checks come from the
92b8afe assessor. Its TG-only diagnostic identity and early-sampling
checks are not used. Instead every record must carry
`diagnostic=scaled_shear_n256`, and N128 records must carry none.
- The reference N128 receipts are hash-pinned to the audited de5b0cc
evidence (commit cba6bca). They are never relabelled as fresh controls.
- The report always includes, for each case:
- the raw signed rate bias `rate/exact - 1` and the interval errors;
- each refined run's signed excess over *each* uniform control (rate
bias, residual, raw pressure rms);
- the N128→N256 signed changes and ratios. The ratio is null when the
N128 value is zero.
- These are descriptive. Passing or failing another floor-controlled
screen is not the conclusion, and a two-resolution ratio is not a
convergence order.
tests/validation/taylor_green/assess_shear_wave.py
+116 -9
View File
@@ -1,4 +1,9 @@
"""Fixed N128 four-case shear-wave contract: U128, U256, constant R128, projected R128.
"""Fixed four-case shear-wave contracts.
Default: N128 (U128, U256, constant R128, projected R128). Opt-in
--diagnostic scaled_shear_n256: N256 (U256, U512, constant R256, projected
R256), same physical rectangle, t* schedule, nu = U N / 100, eps(N), limits
and floors, plus a descriptive comparison with the hash-pinned N128 receipts.
Shared provenance, build, workload and sample-time checks are reused from
assess_projected_pairs (same cases, source pins and geometry); only the
@@ -6,12 +11,23 @@ Taylor-Green-specific checks are replaced by shear-wave metadata and physics.
"""
import argparse
import hashlib
import json
from pathlib import Path
import assess_projected_pairs as tg
import shear_wave as sw
from run_case import SHEAR_WAVE_SHA256
from run_case import SCALED_DIAGNOSTIC, SCALED_SHEAR, SHEAR_WAVE_SHA256
# Audited de5b0cc N128 receipts (evidence commit cba6bca, shear-wave-de5b0cc zip).
REFERENCE_REVISION = "de5b0cc080794d0ce44d37f845aa8e8dbf8c2016"
REFERENCE_RECEIPTS = {
"uniform128": "d4a60fb806d04bfa8098a862cceba51e00b49979a6a250fef17a5fc8e3ea3255",
"uniform256": "e0b58e734724be5f82ee5d6df84395e3181b4e96e4246dd5a574ea23a8d12386",
"constant128": "c70d7cd14b1257a01cc49f8edfcc7685d430e5301e2f2d60ed426936f8b96c51",
"projected128": "980161458b9463af09a8e3178063fea0f4bbeb465329e1d3becdb0f224537b54",
}
ROLES = ("uniform_root", "uniform_fine", "constant", "projected_linear")
SHARED = ("complete_matrix", "no_extra_cases", "no_duplicate_cases", "finite_receipts",
"same_source_and_harness", "clean_a135252_numerical_source",
@@ -33,15 +49,26 @@ def metadata_ok(record, n):
and (record.get("source") or {}).get("shear_wave_sha256") == SHEAR_WAVE_SHA256)
def assess(records, expected_revision=None):
def cases_for(scaled):
factor = 2 if scaled else 1
return tuple((re, n * factor, depth, scheme) for re, n, depth, scheme in tg.CASES)
def assess(records, expected_revision=None, *, scaled=False):
if not isinstance(records, list) or not all(isinstance(r, dict) for r in records):
raise ValueError("records must be a list of receipt objects")
base_checks = tg.assess(records, expected_revision)["checks"]
# The scaled TG assessor supplies N256/512 cases, builds and workload; its
# TG-only diagnostic identity and early-sampling checks are not used here.
base_checks = tg.assess(records, expected_revision,
diagnostic=SCALED_DIAGNOSTIC if scaled else None)["checks"]
checks = {name: base_checks.get(name, False) for name in SHARED}
checks["shear_diagnostic_identity"] = all(
r.get("diagnostic") == (SCALED_SHEAR if scaled else None) for r in records)
cases = cases_for(scaled)
indexed = {}
for r in records:
indexed.setdefault(tuple(r.get(k) for k in ("re_L", "n", "depth", "explosion")), r)
for key in tg.CASES:
for key in cases:
_, n, depth, scheme = key
label = f"{n}/{depth}/{scheme}"
for name in PER_CASE:
@@ -49,32 +76,112 @@ def assess(records, expected_revision=None):
checks[f"{label} shear_metadata"] = key in indexed and metadata_ok(indexed[key], n)
contract_valid = all(checks.values())
result = dict(checks=checks, contract_valid=contract_valid,
cases=[list(k) for k in tg.CASES], expected_revision=expected_revision,
cases=[list(k) for k in cases], expected_revision=expected_revision,
diagnostic=SCALED_SHEAR if scaled else None,
scope="2D periodic Re100 oblique (2,1) shear wave; one phase and one "
"rectangle; no placement robustness or convergence order")
if not contract_valid:
result.update(status="INVALID_SET", passed=False)
return result
try:
physics = sw.assess(indexed[tg.CASES[0]], indexed[tg.CASES[1]],
[indexed[tg.CASES[2]], indexed[tg.CASES[3]]])
physics = sw.assess(indexed[cases[0]], indexed[cases[1]],
[indexed[cases[2]], indexed[cases[3]]])
except (KeyError, ValueError) as error:
result.update(status="INVALID_SET", passed=False, reason=str(error))
return result
result.update(physics=physics, status=physics["status"],
passed=physics["status"] == "PASS")
try:
result["signed"] = signed_summary([indexed[k] for k in cases])
except (KeyError, TypeError, ValueError) as error:
result["signed_unavailable"] = str(error)
return result
def _signed(record):
"""Raw signed quantities of one valid record (None when the fit is unavailable)."""
s = sw.summarize(record)
bias = None if not s["rate_fit_valid"] else s["rate"] / s["exact_rate"] - 1
return dict(n=record["n"], depth=record["depth"], explosion=record["explosion"],
rate_fit_valid=s["rate_fit_valid"], signed_rate_bias=bias,
rate_tstar=s["rate_tstar"], exact_rate_tstar=s["exact_rate_tstar"],
interval_rate_errors=s["interval_rate_errors"],
residual_l2=s["residual_l2"], spurious_pressure_scaled=s["spurious_pressure_scaled"],
spurious_pressure_rms=s["spurious_pressure_rms"],
final_amplitude_error=record["rows"][-1]["amplitude_error"])
def signed_summary(records):
"""Signed bias per case and refined excess over each uniform control."""
rows = dict(zip(ROLES, (_signed(r) for r in records)))
def minus(a, b):
return None if a is None or b is None else a - b
for role in ("constant", "projected_linear"):
for control in ("uniform_root", "uniform_fine"):
rows[role][f"excess_over_{control}"] = {
key: minus(rows[role][key], rows[control][key])
for key in ("signed_rate_bias", "residual_l2", "spurious_pressure_rms")}
return rows
def load_reference(paths):
"""The four audited N128 receipts verbatim; never relabelled as fresh controls."""
blobs = [Path(path).read_bytes() for path in paths]
hashes = {hashlib.sha256(blob).hexdigest() for blob in blobs}
if len(blobs) != 4 or hashes != set(REFERENCE_RECEIPTS.values()):
raise ValueError("N128 receipt hashes differ from the audited de5b0cc evidence")
return [json.loads(blob) for blob in blobs]
def resolution_trend(new_result, reference, reference_revision=REFERENCE_REVISION):
"""Descriptive N128 -> N256 signed changes; never an acceptance gate or an order."""
old_result = assess(reference, reference_revision)
if "signed" not in old_result or "signed" not in new_result:
return dict(valid=False, reason="N128 reference or N256 set lacks a valid signed summary")
out = {}
for role in ROLES:
old, new = old_result["signed"][role], new_result["signed"][role]
out[role] = {}
for key in ("signed_rate_bias", "residual_l2", "spurious_pressure_rms",
*(f"excess_over_{c}" for c in ("uniform_root", "uniform_fine"))):
if key not in old:
continue
pairs = old[key].items() if isinstance(old[key], dict) else [(None, old[key])]
for sub, low in pairs:
high = new[key][sub] if sub else new[key]
ok = low is not None and high is not None
out[role][key if sub is None else f"{key}.{sub}"] = dict(
n128=low, n256=high, signed_change=high - low if ok else None,
n256_to_n128=high / low if ok and low else None)
return dict(valid=True, reference_revision=reference_revision,
reference_receipt_sha256=REFERENCE_RECEIPTS,
reference_status=old_result["status"], changes=out,
scope="same-placement two-resolution trend; not a convergence order, "
"placement robustness or redesign decision")
if __name__ == "__main__":
p = argparse.ArgumentParser(description=__doc__)
p.add_argument("files", type=Path, nargs=4)
p.add_argument("--output", type=Path, required=True)
p.add_argument("--expected-revision", required=True)
p.add_argument("--diagnostic", choices=[SCALED_SHEAR])
p.add_argument("--reference-files", type=Path, nargs=4)
args = p.parse_args()
if args.output.exists():
p.error("Output already exists; choose a new path to retain prior evidence")
result = assess([json.loads(path.read_text()) for path in args.files], args.expected_revision)
if bool(args.diagnostic) != bool(args.reference_files):
p.error("--diagnostic scaled_shear_n256 requires --reference-files (N128 receipts)")
result = assess([json.loads(path.read_text()) for path in args.files], args.expected_revision,
scaled=bool(args.diagnostic))
if args.diagnostic:
try:
reference = load_reference(args.reference_files)
except ValueError as exc:
p.error(str(exc))
result["resolution_trend"] = resolution_trend(result, reference)
text = json.dumps(result, indent=2, allow_nan=False) + "\n"
args.output.write_text(text)
print(text, end="")
tests/validation/taylor_green/run_case.py
+13 -6
View File
@@ -20,6 +20,7 @@ REFERENCE_HARNESS = "11901a7f77557a901e017a171eeae1758d584fa5"
INITIALIZATION = "physical FP64 equilibrium G at actual cell centers; rho=1+3p"
SHEAR_WAVE_SHA256 = "81d5c8ee60864372887cddf9239580c82fb6750d956038f1e3ee9de61aa4e877"
SCALED_DIAGNOSTIC = "scaled_n256"
SCALED_SHEAR = "scaled_shear_n256" # mutually exclusive with scaled_n256 (one flow each)
def flow_functions(flow="taylor_green"):
@@ -181,8 +182,9 @@ def main():
p.add_argument("--depth", type=int, choices=[1, 2, 3], required=True)
p.add_argument("--output", type=Path, required=True)
p.add_argument("--explosion", choices=["constant", "projected_linear"], default="constant")
p.add_argument("--diagnostic", choices=[SCALED_DIAGNOSTIC],
help="explicit four-case N256/fine512 same-placement diagnostic")
p.add_argument("--diagnostic", choices=[SCALED_DIAGNOSTIC, SCALED_SHEAR],
help="explicit four-case N256/fine512 same-placement diagnostic "
"(scaled_n256: Taylor-Green; scaled_shear_n256: shear wave)")
p.add_argument("--early-pressure", action="store_true",
help="also save 65 scalar pressure samples at t*=k*.001875, k=0..64")
p.add_argument("--flow", choices=["taylor_green", "shear_wave"], default="taylor_green")
@@ -197,12 +199,17 @@ def main():
if args.diagnostic:
cases = {(256, 1, "constant"), (512, 1, "constant"),
(256, 2, "constant"), (256, 2, "projected_linear")}
if args.flow != "taylor_green":
p.error("--diagnostic scaled_n256 requires --flow taylor_green")
if args.re != 100 or (args.n, args.depth, args.explosion) not in cases or not args.early_pressure:
flow = "shear_wave" if args.diagnostic == SCALED_SHEAR else "taylor_green"
if args.flow != flow:
p.error(f"--diagnostic {args.diagnostic} requires --flow {flow}")
if args.diagnostic == SCALED_DIAGNOSTIC and (
args.re != 100 or (args.n, args.depth, args.explosion) not in cases
or not args.early_pressure):
p.error("scaled_n256 requires one of its four Re100 cases and --early-pressure")
if args.diagnostic == SCALED_SHEAR and (args.n, args.depth, args.explosion) not in cases:
p.error("scaled_shear_n256 requires one of its four Re100 cases")
elif args.n == 512:
p.error("N512 requires --diagnostic scaled_n256")
p.error("N512 requires --diagnostic scaled_n256 or scaled_shear_n256")
selected = os.environ.get("CUDA_VISIBLE_DEVICES", "")
if not args.dry_run and (not args.run_gpu or not selected or "," in selected or selected == "-1"):
p.error("--run-gpu and exactly one explicit CUDA_VISIBLE_DEVICES are required")
tests/validation/taylor_green/test_shear_wave.py
+92 -3
View File
@@ -4,8 +4,13 @@ Run from this directory with PYTHONPATH=<checkout>/src (same as test_oracle.py).
"""
import hashlib
import io
import json
import math
import sys
import tempfile
from contextlib import redirect_stdout
from unittest.mock import patch
import unittest
from pathlib import Path
@@ -361,7 +366,7 @@ class RunnerSampleTests(unittest.TestCase):
self.assertEqual(row["leaf_cells"], N * N)
def receipt(n, depth, explosion="constant"):
def receipt(n, depth, explosion="constant", diagnostic=None):
"""Synthetic receipt satisfying every contract field of the fixed four-case set."""
import assess_projected_pairs as tg
steps = run_case.sample_steps(n)[0]
@@ -369,7 +374,7 @@ def receipt(n, depth, explosion="constant"):
for row, step in zip(rows, steps):
row.update(step=step, tstar=step * U / n,
spurious_pressure_scaled_by_level=[row["spurious_pressure_scaled"]] * depth)
cells = [n * n] if depth == 1 else [8704, 30720]
cells = [n * n] if depth == 1 else [17 * n * n // 32, 15 * n * n // 8]
hops = [16 * n * 2**level for level in range(depth)]
builds = []
for level in range(depth):
@@ -406,7 +411,8 @@ def receipt(n, depth, explosion="constant"):
active_collision_updates=sum(c * h for c, h in zip(cells, hops))),
actual_build_sidecars=builds, rows=rows, completed=True,
flow="shear_wave", mode=list(sw.MODE), polarization=sw.E.tolist(),
angle_deg=sw.ANGLE_DEG, phase=sw.PHI, eps=sw.eps(n))
angle_deg=sw.ANGLE_DEG, phase=sw.PHI, eps=sw.eps(n),
**({"diagnostic": diagnostic} if diagnostic else {}))
class ContractTests(unittest.TestCase):
@@ -448,6 +454,89 @@ class ContractTests(unittest.TestCase):
self.assertEqual(a(self.records, "wrong-revision")["status"], "INVALID_SET")
def scaled_set(rate_error=0.0):
d = run_case.SCALED_SHEAR
out = [receipt(256, 1, diagnostic=d), receipt(512, 1, diagnostic=d),
receipt(256, 2, diagnostic=d), receipt(256, 2, "projected_linear", diagnostic=d)]
if rate_error:
out[3]["rows"] = rows_for(256, rate_error=rate_error)
for row, step in zip(out[3]["rows"], run_case.sample_steps(256)[0]):
row.update(step=step, tstar=step * U / 256,
spurious_pressure_scaled_by_level=[row["spurious_pressure_scaled"]] * 2)
return out
class ScaledContractTests(unittest.TestCase):
def setUp(self):
import assess_shear_wave
self.mod = assess_shear_wave
def test_scaled_contract_and_signed_excess(self):
result = self.mod.assess(scaled_set(1.5e-3), "r", scaled=True)
self.assertTrue(result["contract_valid"], [k for k, v in result["checks"].items() if not v])
self.assertEqual([c[1] for c in result["cases"]], [256, 512, 256, 256])
projected = result["signed"]["projected_linear"]
self.assertAlmostEqual(projected["signed_rate_bias"], 1.5e-3, delta=1e-9)
self.assertAlmostEqual(projected["excess_over_uniform_root"]["signed_rate_bias"], 1.5e-3,
delta=1e-9)
self.assertIn("excess_over_uniform_fine", projected)
def test_scaled_and_default_sets_do_not_mix(self):
n128 = [receipt(128, 1), receipt(256, 1), receipt(128, 2), receipt(128, 2, "projected_linear")]
self.assertEqual(self.mod.assess(n128, "r", scaled=True)["status"], "INVALID_SET")
self.assertEqual(self.mod.assess(scaled_set(), "r")["status"], "INVALID_SET")
unlabeled = scaled_set()
unlabeled[0].pop("diagnostic")
self.assertEqual(self.mod.assess(unlabeled, "r", scaled=True)["status"], "INVALID_SET")
tg_label = scaled_set()
tg_label[2]["diagnostic"] = run_case.SCALED_DIAGNOSTIC
self.assertEqual(self.mod.assess(tg_label, "r", scaled=True)["status"], "INVALID_SET")
def test_resolution_trend_and_reference_pin(self):
reference = [receipt(128, 1), receipt(256, 1), receipt(128, 2),
receipt(128, 2, "projected_linear")]
new = self.mod.assess(scaled_set(1.5e-3), "r", scaled=True)
trend = self.mod.resolution_trend(new, reference, "r")
self.assertTrue(trend["valid"])
change = trend["changes"]["projected_linear"]["signed_rate_bias"]
self.assertAlmostEqual(change["signed_change"], 1.5e-3, delta=1e-9)
self.assertIn("excess_over_uniform_fine.residual_l2", trend["changes"]["constant"])
with tempfile.TemporaryDirectory() as tmp:
paths = []
for i, r in enumerate(reference):
paths.append(Path(tmp) / f"{i}.json")
paths[-1].write_text(json.dumps(r))
with self.assertRaises(ValueError):
self.mod.load_reference(paths)
def test_scaled_runner_options(self):
repo = Path(__file__).resolve().parents[3]
with tempfile.TemporaryDirectory() as tmp:
base = ["run_case.py", "--repo", str(repo), "--expected-revision", "f" * 40,
"--output", str(Path(tmp) / "case.json"), "--dry-run", "--re", "100"]
shear = ["--flow", "shear_wave"]
scaled = ["--diagnostic", run_case.SCALED_SHEAR]
for extra, valid in (
(["--n", "512", "--depth", "1"] + shear + scaled, True),
(["--n", "256", "--depth", "2", "--explosion", "projected_linear"] + shear + scaled, True),
(["--n", "512", "--depth", "1"] + shear, False),
(["--n", "128", "--depth", "2"] + shear + scaled, False),
(["--n", "512", "--depth", "1"] + scaled, False), # needs shear flow
(["--n", "512", "--depth", "1", "--early-pressure", "--diagnostic",
run_case.SCALED_DIAGNOSTIC] + shear, False), # mutually exclusive
(["--n", "256", "--depth", "2", "--early-pressure"] + shear + scaled, False)):
argv = base + extra
with self.subTest(argv=argv), patch.object(sys, "argv", argv), \
patch.object(run_case, "verify_source", return_value={}), \
patch.object(run_case, "runtime_source", return_value={}), \
redirect_stdout(io.StringIO()), patch("sys.stderr", io.StringIO()):
if valid:
self.assertEqual(run_case.main(), 0)
else:
with self.assertRaises(SystemExit):
run_case.main()
class DefaultPathTests(unittest.TestCase):
def test_frozen_oracle_is_byte_identical(self):
digest = hashlib.sha256((HERE / "oracle.py").read_bytes()).hexdigest()