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

test: depth-2 twin of the nested shear preset (twin_shear_n128)

Browse Source 
Test-only opt-in. New shear_twin.py: same two L0->L1 boxes as
shear_nested, no L2; leaves [7936,33792], hops [2048,4096], work
154664960. Four fresh cases U128/U256/constant R128 d2/projected R128 d2;
physics is the unchanged depth-2/2N shear_wave.assess. run_case:
--diagnostic twin_shear_n128 with shear_twin.py+shear_nested.py pins, no
snapshot. assess_shear_wave: nested contract made preset-generic (PRESETS)
with nested behaviour unchanged. Labels: twin vs nested = effect of adding
this nested refinement (not abstract depth). shear_wave.py, shear_nested.py
and oracle.py byte-identical.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
ClaudeandClaude Opus 5.5 committed 2026-10-19 02:52:20 +00:00
1 parent 335624c27b
commit 02bde0f751
5 files changed
+304 -41
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

tests/validation/taylor_green/assess_shear_wave.py
+55 -30
View File
@@ -19,8 +19,9 @@ from pathlib import Path
import assess_projected_pairs as tg
import shear_wave as sw
import shear_nested as nested
import shear_twin as twin
from run_case import (NESTED_SHEAR, SCALED_DIAGNOSTIC, SCALED_SHEAR, SHEAR_NESTED_SHA256,
SHEAR_WAVE_SHA256, sample_steps)
SHEAR_TWIN_SHA256, SHEAR_WAVE_SHA256, TWIN_SHEAR, sample_steps)
# Audited de5b0cc N128 receipts (evidence commit cba6bca, shear-wave-de5b0cc zip).
REFERENCE_REVISION = "de5b0cc080794d0ce44d37f845aa8e8dbf8c2016"
@@ -120,15 +121,38 @@ CASE_AGNOSTIC = ("finite_receipts", "same_source_and_harness", "clean_a135252_nu
"same_recorded_runtime", "same_explicit_device", "same_early_sampling")
def _nested_case_ok(r, n, depth):
"""Per-case identity of the fixed nested preset (scales, builds, patch, work, pins)."""
# Fixed multi-box presets sharing one contract; only geometry, controls, pins differ.
PRESETS = {
NESTED_SHEAR: dict(
module=nested, physics=nested.assess, snapshot=True,
pins={"shear_nested_sha256": SHEAR_NESTED_SHA256},
finest_control="uniform 4N (U512)",
scope="2D periodic Re100 oblique (2,1) shear wave; fixed three-level/two-region "
"placement; changes placement and refined volume as well as depth, so no "
"isolated-level cost or resolution trend"),
TWIN_SHEAR: dict(
module=twin, physics=sw.assess, snapshot=False,
pins={"shear_nested_sha256": SHEAR_NESTED_SHA256,
"shear_twin_sha256": SHEAR_TWIN_SHA256},
finest_control="uniform 2N (U256)",
scope="2D periodic Re100 oblique (2,1) shear wave; depth-2 twin of the nested "
"preset (same two L0->L1 boxes, no L2). Twin vs nested is the effect of adding "
"this nested refinement; it also changes finest spacing and refined volume, so it "
"does not isolate abstract depth"),
}
def _nested_case_ok(r, n, depth, diagnostic=NESTED_SHEAR):
"""Per-case identity of a fixed multi-box preset (scales, builds, patch, work, pins)."""
preset = PRESETS[diagnostic]
geometry = preset["module"]
runtime = r.get("runtime_source", {})
package = Path(runtime.get("checkout", "")) / "src/CelerisLab"
cells = [n * n] if depth == 1 else nested.LEAVES
cells = [n * n] if depth == 1 else geometry.LEAVES
hops = [16 * n * 2**level for level in range(depth)]
work = r.get("workload", {})
snapshot = r.get("final_snapshot")
expected_patch = [] if depth == 1 else nested.REFINEMENTS
expected_patch = [] if depth == 1 else geometry.REFINEMENTS
return {
"complete_times": r.get("completed") is True and tg.at_steps(
r.get("rows", []), sample_steps(n)[0], n),
@@ -152,11 +176,11 @@ def _nested_case_ok(r, n, depth):
and work.get("active_collision_updates") == sum(
c * h for c, h in zip(cells, hops))),
"shear_metadata": metadata_ok(r, n),
"nested_module_pin": (r.get("source") or {}).get("shear_nested_sha256")
== SHEAR_NESTED_SHA256,
# Optional, projected depth-3 only; bounded and hashed when present.
"nested_module_pin": all((r.get("source") or {}).get(key) == pin
for key, pin in preset["pins"].items()),
# Optional, nested projected depth-3 only; bounded and hashed when present.
"snapshot": snapshot is None or (
depth == nested.DEPTH and r.get("explosion") == "projected_linear"
preset["snapshot"] and depth == nested.DEPTH and r.get("explosion") == "projected_linear"
and snapshot.get("kept") is True and isinstance(snapshot.get("sha256"), str)
and len(snapshot["sha256"]) == 64
and 0 < snapshot.get("bytes", 0) <= nested.SNAPSHOT_MAX_BYTES
@@ -165,8 +189,11 @@ def _nested_case_ok(r, n, depth):
}
def assess_nested(records, expected_revision=None):
"""Fixed nested_shear_n128 four-case contract (U128, U512, constant/projected R128 d3)."""
def assess_nested(records, expected_revision=None, diagnostic=NESTED_SHEAR):
"""Fixed four-case preset contract: nested_shear_n128 (U128, U512, constant/projected
R128 depth 3) or twin_shear_n128 (U128, U256, constant/projected R128 depth 2)."""
preset = PRESETS[diagnostic]
geometry = preset["module"]
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"]
@@ -175,38 +202,36 @@ def assess_nested(records, expected_revision=None):
for r in records:
key = tuple(r.get(k) for k in ("re_L", "n", "depth", "explosion"))
(duplicates.append(key) if key in indexed else indexed.__setitem__(key, r))
checks.update(complete_matrix=all(k in indexed for k in nested.CASES),
no_extra_cases=set(indexed) <= set(nested.CASES),
checks.update(complete_matrix=all(k in indexed for k in geometry.CASES),
no_extra_cases=set(indexed) <= set(geometry.CASES),
no_duplicate_cases=not duplicates,
shear_diagnostic_identity=all(r.get("diagnostic") == NESTED_SHEAR
shear_diagnostic_identity=all(r.get("diagnostic") == diagnostic
for r in records))
for key in nested.CASES:
for key in geometry.CASES:
_, n, depth, scheme = key
label = f"{n}/{depth}/{scheme}"
case = _nested_case_ok(indexed[key], n, depth) if key in indexed else {}
case = _nested_case_ok(indexed[key], n, depth, diagnostic) if key in indexed else {}
for name in ("complete_times", "scales", "actual_builds", "reference", "runtime_source",
"patch", "workload", "shear_metadata", "nested_module_pin", "snapshot"):
checks[f"{label} {name}"] = bool(case.get(name, False))
contract_valid = all(checks.values())
result = dict(checks=checks, contract_valid=contract_valid, diagnostic=NESTED_SHEAR,
cases=[list(k) for k in nested.CASES], expected_revision=expected_revision,
finest_control="uniform 4N (U512)",
scope="2D periodic Re100 oblique (2,1) shear wave; fixed three-level/two-region "
"placement; changes placement and refined volume as well as depth, so no "
"isolated-level cost or resolution trend")
result = dict(checks=checks, contract_valid=contract_valid, diagnostic=diagnostic,
cases=[list(k) for k in geometry.CASES], expected_revision=expected_revision,
finest_control=preset["finest_control"], scope=preset["scope"])
if not contract_valid:
result.update(status="INVALID_SET", passed=False)
return result
try:
physics = nested.assess(indexed[nested.CASES[0]], indexed[nested.CASES[1]],
[indexed[nested.CASES[2]], indexed[nested.CASES[3]]])
cases = geometry.CASES
physics = preset["physics"](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:
# Role "uniform_fine" here is the 4N finest control.
result["signed"] = signed_summary([indexed[k] for k in nested.CASES])
# Role "uniform_fine" is the preset's finest control (4N nested, 2N twin).
result["signed"] = signed_summary([indexed[k] for k in geometry.CASES])
except (KeyError, TypeError, ValueError) as error:
result["signed_unavailable"] = str(error)
return result
@@ -281,16 +306,16 @@ if __name__ == "__main__":
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, NESTED_SHEAR])
p.add_argument("--diagnostic", choices=[SCALED_SHEAR, NESTED_SHEAR, TWIN_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")
if args.diagnostic == NESTED_SHEAR:
if args.diagnostic in PRESETS:
if args.reference_files:
p.error("nested_shear_n128 takes no reference files (no resolution trend)")
p.error(f"{args.diagnostic} takes no reference files (no resolution trend)")
result = assess_nested([json.loads(path.read_text()) for path in args.files],
args.expected_revision)
args.expected_revision, args.diagnostic)
text = json.dumps(result, indent=2, allow_nan=False) + "\n"
args.output.write_text(text)
print(text, end="")