"""FRESH-V2 qualification: independence audit, per-variant fresh bounds, seal before query GT.""" import copy import importlib.util import json from pathlib import Path import pytest import torch from test_geometry_carrier_evaluation import fixture as evaluation_fixture from mcss.mechanism_pilot.rgbd_bounds_carrier import make_carrier from mcss.mechanism_pilot.rgbd_evidence_carrier import RGBDContext from mcss.mechanism_pilot.rgbd_fresh_v2_evaluation import ( FreshDepthBoundsSceneData, FreshRGBDSceneData, FreshTrimmedDepthBoundsSceneData, evaluate_model, fresh_scene_data, ) def script(name): path = Path(__file__).parents[1] / "scripts" / name spec = importlib.util.spec_from_file_location(name.replace(".py", ""), path) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module def fresh_fixture(root): manifest, _ = evaluation_fixture(root) for record in manifest["scenes"]: if record["split"] == "DEV": record.update(split="FRESH_QUALIFICATION", historically_exposed=False) record["independence_verified"] = True manifest["fresh_status"] = "VERIFIED" return manifest def test_independence_audit_accepts_clean_and_rejects_every_overlap(): audit = script("prepare_rgbd_fresh_v2_qualification.py").independence_audit v7 = {"scenes": [{"scene_id": "ai_001_001", "physical_scene_id": "p1"}]} names = ("ai_001_002", "ai_002_001", "ai_002_002", "ai_002_003", "ai_003_001", "ai_004_001") partitions = [ { "scene_name": s, "official_split": "train", "protocol_partition": "train", "previously_observed": "False", } for s in names ] def record(sid, physical): return { "scene_id": sid, "split": "FRESH_QUALIFICATION", "physical_scene_id": physical, "historically_exposed": False, } clean = [record("ai_002_001", "p2"), record("ai_002_002", "p3"), record("ai_003_001", "p4")] assert all(a["verified"] for a in audit(clean, v7, set(), set(), partitions, set())) broken = [ ([record("ai_002_001", "p1")], set(), set()), # physical identity shared with V7 (clean, set(), {"p3"}), # physical identity shared with an earlier cohort (clean, {"ai_003_001"}, set()), # used or considered by an earlier experiment (clean + [record("ai_002_003", "p5")], set(), set()), # three scenes in one volume ([record("ai_001_002", "p6")], set(), set()), # volume shared with V7 TRAIN/DEV ([{**record("ai_002_001", "p2"), "historically_exposed": True}], set(), set()), ] for records, earlier, physical in broken: with pytest.raises(PermissionError): audit(records, v7, earlier, physical, partitions, set()) with pytest.raises(PermissionError): audit(clean, v7, set(), set(), partitions, {"ai_003_001"}) for key, value in (("previously_observed", "True"), ("official_split", "val")): changed = copy.deepcopy(partitions) changed[1][key] = value with pytest.raises(PermissionError): audit(clean, v7, set(), set(), changed, set()) with pytest.raises(PermissionError): audit([record("ai_004_001", "p7")], v7, set(), set(), partitions[:-1], set()) def test_fresh_loaders_follow_each_v7_bounds_rule_and_never_serve_targets(tmp_path): manifest = fresh_fixture(tmp_path) record = next(r for r in manifest["scenes"] if r["split"] == "FRESH_QUALIFICATION") assert fresh_scene_data("C0") is FreshRGBDSceneData assert fresh_scene_data("C1") is FreshDepthBoundsSceneData assert fresh_scene_data("C2") is FreshTrimmedDepthBoundsSceneData frozen = FreshRGBDSceneData(record, manifest, tmp_path, "cpu", []) access = [] depth = FreshDepthBoundsSceneData(record, manifest, tmp_path, "cpu", access) context = depth.context("A") assert isinstance(context, RGBDContext) and context.depths.shape[0] == len(context) assert not torch.equal(frozen.bounds["A"], depth.bounds["A"]) assert all( e["purpose"] == "CONTEXT_ONLY_RGBD" for e in access if "depth" in e.get("channels", []) ) with pytest.raises(PermissionError): depth.target(0) with pytest.raises(PermissionError): blocked = {**manifest, "fresh_status": "BLOCKED"} FreshDepthBoundsSceneData(record, blocked, tmp_path, "cpu", []) train = next(r for r in manifest["scenes"] if r["split"] == "TRAIN") with pytest.raises(PermissionError): FreshTrimmedDepthBoundsSceneData(train, manifest, tmp_path, "cpu", []) def test_fresh_v2_evaluation_seals_every_state_before_query_depth(tmp_path): torch.set_num_threads(1) manifest = fresh_fixture(tmp_path) records = {r["scene_id"]: r for r in manifest["scenes"]} boxes = {} for variant in ("C0", "C1"): carrier = make_carrier(variant, 5, "cpu") with torch.no_grad(): carrier.depth_weight.fill_(0.05) out = tmp_path / f"evaluation_{variant}" result = evaluate_model( carrier, manifest, tmp_path, out, variant, 5, 100, "cpu", diagnostics=True ) assert len(result["query_rows"]) == 40 access = json.loads((out / "GT_access.json").read_text()) marker = next( i for i, e in enumerate(access) if e.get("event") == "ALL_FRESH_STATES_SEALED" ) early = [e for e in access[:marker] if "depth" in e.get("channels", [])] assert early for event in early: roles = records[event["scene_id"]]["roles"] assert event["purpose"] == "CONTEXT_ONLY_RGBD" assert event["frame_id"] in set(roles["context_a"]) | set(roles["context_b"]) metadata = json.loads((out / "state_hashes.json").read_text()) boxes[variant] = {k: v["bounds"] for k, v in metadata.items()} assert any(boxes["C0"][k] != boxes["C1"][k] for k in boxes["C0"]) dev = copy.deepcopy(manifest) for r in dev["scenes"]: if r["split"] == "FRESH_QUALIFICATION": r["split"] = "DEV" with pytest.raises(PermissionError): evaluate_model(carrier, dev, tmp_path, tmp_path / "blocked", "C1", 5, 100, "cpu") def test_volume_bootstrap_resamples_volumes_not_scenes(): report = script("report_rgbd_fresh_v2_qualification.py") result = report.volume_bootstrap( {"ai_001_001": 1.0, "ai_001_002": 3.0, "ai_002_001": -1.0}, draws=2000 ) assert result["n_volumes"] == 2 and result["volumes_positive"] == 1 assert result["mean"] == pytest.approx(0.5) assert result["ci95"][0] == pytest.approx(-1.0) and result["ci95"][1] == pytest.approx(2.0) def test_report_requires_all_four_gates(tmp_path): from test_geometry_carrier_statistics import fixture, region_fixture data = fixture() v7 = tmp_path / "v7" v7.mkdir() (v7 / "terminal_summary.txt").write_text("BOUNDS_GAIN=0.1\nINTERPRETATION_BRANCH=A\n") values = { "scene_split": data[3], "qualification_contract": {"seeds": data[4]["seeds"], "v7_root": str(v7)}, "raw/fresh_query_results": data[0], "raw/fresh_context_results": data[1], "raw/fresh_matched_results": data[2], "raw/fresh_region_results": region_fixture(data), "raw/fresh_static_matched_results": data[4]["static_matched_rows"], "audit/fresh_state_use": data[4]["state_use_audit"], "training_curves": {"training": [], "dev": []}, "cost_analysis": {"variants": {v: {"training_seconds": 1.0} for v in ("C0", "C1")}}, } for name, value in values.items(): path = tmp_path / (name + ".json") path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(value)) artifacts = script("analyze_rgbd_fresh_v2_qualification.py").run(tmp_path) assert artifacts["ray_hit_fractions"]["descriptive_only"] scenes = sorted({r["scene_id"] for r in data[0]}) def cell(label): return { "reference_label": label, "vs_REF_TRAIN_ABSREL_OPTIMAL": { "mean": 0.1, "ci95": [0.05, 0.15], "improved_tied_worse": [len(scenes), 0, 0], "per_scene": {s: 0.1 for s in scenes}, }, } report = script("report_rgbd_fresh_v2_qualification.py") for label in ("NOT_DISTINGUISHABLE", "ABOVE"): reference = { "summary": {"CARRIER_REFERENCE_STATUS": "X"}, "carriers": {"V7_C1": {"RAW": {"depth_absrel": cell(label)}}}, } (tmp_path / "reference_results.json").write_text(json.dumps(reference)) fields = report.run(tmp_path) others = ( fields["BOUNDS_STATUS_FRESH"], fields["SCENE_SPECIFICITY_STATUS_FRESH"], fields["STATIC_STATUS_FRESH"], ) expected = label == "ABOVE" and all(g == "SUPPORTED" for g in others) assert fields["REFERENCE_STATUS_FRESH"] == ( "SUPPORTED" if label == "ABOVE" else "NOT_ESTABLISHED" ) assert (fields["FRESH_QUALIFICATION_STATUS"] == "QUALIFIED") == expected assert fields["DYNAMIC_TTT_NEXT_STAGE_ALLOWED"] == expected