"""Synthetic matched carrier DEV gates, never media or model predictions.""" import copy import json import pytest from mcss.mechanism_pilot.geometry_carrier_statistics import METHODS, METRICS, analyze def fixture(): ids, seeds = [f"s{i}" for i in range(8)], [20260928, 20260929] manifest = { "scenes": [ { "scene_id": s, "split": "DEV", "roles": {"primary_query": [8, 9], "context_a": [0, 1, 2], "context_b": [0, 3, 4]}, } for s in ids ], "fresh_status": "BLOCKED_INDEPENDENCE_UNRESOLVED", } query, context, matched = [], [], [] for scene in ids: for seed in seeds: for role in ("A", "B"): for variant in ("C0", "C1"): value = ( 0.4 if variant == "C0" else 0.39 ) # Below old .02 threshold deliberately. for fid in (8, 9): base = dict( scene_id=scene, seed=seed, role=role, variant=variant, query_id=fid ) for method in METHODS: error = value + (0 if method == "direct" else 0.1) query.append( { **base, "method": method, **{m: error for m in METRICS}, "coverage": 1.0, "opacity": 0.8, "ray_hitfraction": 1.0, } ) matched.append( { **base, "method": "direct", "depth_absrel": value, "normalized_depth_absrel": value, "common_valid_count": 9, "total_gt_valid": 9, "common_mask_hash": "a" * 64, } ) for fid in [0, 1, 2] if role == "A" else [0, 3, 4]: context.append( dict( scene_id=scene, seed=seed, role=role, variant=variant, method="direct", frame_id=fid, **{m: 0.1 for m in METRICS}, ) ) audit = dict( status="PASS", shared_state_multiple_queries=True, query_after_state_seal=True, recipient_camera_preserved=True, ) return query, context, matched, manifest, dict(seeds=seeds, state_use_audit=audit, draws=100) def run(data): q, c, m, manifest, kwargs = data return analyze(q, c, m, manifest, **kwargs) def test_positive_gain_under_old_threshold_and_fresh_remains_closed(): result = run(fixture()) q = result["qualification_results"] assert q["SURFACE_TRAINING_STATUS"] == "SUPPORTED" assert q["STATIC_DEV_STATUS"] == "SUPPORTED" assert result["static_results"]["surface_gain"]["mean"] == pytest.approx(0.01) assert not q["FRESH_QUALIFICATION_OPENED"] and not q["DYNAMIC_TTT_NEXT_STAGE_ALLOWED"] assert q["FINAL_STATIC_STATUS"] == "NOT_ESTABLISHED" assert q["SEED_ROBUSTNESS"] == "DIRECTION_REPLICATED" def test_no_hit_scene_kept_in_overall_and_common_mask_null(): data = fixture() for row in data[0]: if row["scene_id"] == "s0": row.update(coverage=0.0, ray_hitfraction=0.0) for row in data[2]: if row["scene_id"] == "s0": row.update(common_valid_count=0, depth_absrel=None, normalized_depth_absrel=None) result = run(data) state = result["state_use_results"] assert state["opacity_artifact_gate"] assert state["no_hit_scenes"] == ["s0"] assert state["common_mask_gains"]["depth_absrel"]["scene_coverage"] == 7 / 8 assert state["common_mask_gains"]["depth_absrel"]["per_scene"]["s0"] is None assert result["static_results"]["surface_gain"]["n_scenes"] == 8 def test_insufficient_common_coverage_or_normalized_gain_blocks_support(): data = fixture() for row in data[2]: if row["variant"] == "C1": row["normalized_depth_absrel"] = 0.4 result = run(data) assert not result["state_use_results"]["opacity_artifact_gate"] assert result["qualification_results"]["SURFACE_TRAINING_STATUS"] == "PARTIAL" data = fixture() for row in data[2]: if row["scene_id"] in ("s0", "s1", "s2"): row.update(common_valid_count=0, depth_absrel=None, normalized_depth_absrel=None) assert not run(data)["state_use_results"]["opacity_artifact_gate"] def test_seed_average_not_pseudo_replication_and_negative_seed_reported(): data = fixture() for row in data[0]: if row["variant"] == "C1" and row["seed"] == 20260929: row["depth_absrel"] += 0.03 result = run(data) assert result["static_results"]["surface_gain"]["n_scenes"] == 8 assert result["static_results"]["surface_gain"]["mean"] == pytest.approx(-0.005) assert result["qualification_results"]["SURFACE_TRAINING_STATUS"] == "HARMFUL" assert result["qualification_results"]["SEED_ROBUSTNESS"] == "NOT_ESTABLISHED" @pytest.mark.parametrize("fault", ["missing", "duplicate", "train", "mask", "hit"]) def test_roster_and_common_mask_fail_closed(fault): data = fixture() if fault == "missing": data[0].pop() if fault == "duplicate": data[0].append(copy.deepcopy(data[0][0])) if fault == "train": data[3]["scenes"][0]["split"] = "TRAIN" if fault == "mask": data[2][0]["common_mask_hash"] = "b" * 64 if fault == "hit": for row in data[0]: if row["variant"] == "C1": row["ray_hitfraction"] = 0.8 with pytest.raises((ValueError, PermissionError)): run(data) def test_state_integrity_required_and_raw_roundtrip_deterministic(): data = fixture() first = run(data) restored = json.loads(json.dumps(data)) assert run(restored) == first data[4]["state_use_audit"]["shared_state_multiple_queries"] = False assert run(data)["qualification_results"]["STATIC_DEV_STATUS"] == "PARTIAL" def region_fixture(data): output = [] for row in data[0]: if row["method"] != "direct": continue for region, count in [ ("OBS0", 0 if row["scene_id"] == "s0" else 3), ("OBS1", 6 if row["scene_id"] == "s0" else 3), ("OBS2PLUS", 3), ]: output.append( { **{k: row[k] for k in ("variant", "seed", "scene_id", "role", "query_id")}, "region": region, "pixel_count": count, "total_valid": 9, "absrel_sum": count * row["depth_absrel"], "absrel_mean": row["depth_absrel"] if count else None, } ) return output def test_region_null_and_partition_contributions(): from mcss.mechanism_pilot.geometry_carrier_statistics import analyze_regions data = fixture() regions = region_fixture(data) result = analyze_regions(regions, data[3], seeds=data[4]["seeds"], draws=100) assert result["groups"]["C0"]["OBS0"]["conditional_absrel"]["empty_mask_scene_count"] == 1 assert result["gains"]["OBS2PLUS"]["mean"] == pytest.approx(0.01) assert sum( result["groups"]["C0"][r]["additive_error_contribution"]["mean"] for r in ("OBS0", "OBS1", "OBS2PLUS") ) == pytest.approx(0.4) regions[0]["pixel_count"] = 1 with pytest.raises(ValueError): analyze_regions(regions, data[3], seeds=data[4]["seeds"], draws=100) def script(name): import importlib.util from pathlib import Path path = Path(__file__).parents[1] / "scripts" / name spec = importlib.util.spec_from_file_location(name, path) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module def test_cli_raw_only_eight_figures_report_and_gzip_reproduction(tmp_path): import gzip data = fixture() values = { "scene_split": data[3], "training_contract": {"seeds": data[4]["seeds"]}, "raw/dev_query_results": data[0], "raw/dev_context_results": data[1], "raw/dev_matched_results": data[2], "raw/dev_region_results": region_fixture(data), "audit/dev_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)) cli = script("analyze_geometry_carrier.py") result = cli.run(tmp_path) assert len(list((tmp_path / "figures").glob("*.png"))) == 8 report = script("report_geometry_carrier.py") fields = report.run(tmp_path) assert not fields["FRESH_QUALIFICATION_OPENED"] assert not fields["DYNAMIC_TTT_NEXT_STAGE_ALLOWED"] assert fields["TEST_TIME_INPUT"] == "RGB+CAMERA" text = (tmp_path / "README.md").read_text() assert all("**" + str(i) + ". " in text for i in range(1, 15)) for p in (tmp_path / "raw").glob("*.json"): compressed = p.with_suffix(".json.gz") compressed.write_bytes(gzip.compress(p.read_bytes(), mtime=0)) p.unlink() result2 = cli.run(tmp_path, tmp_path / "reproduced") assert result == result2 audit = script("audit_geometry_carrier_statistics.py") evidence = audit.audit(tmp_path) assert evidence["status"] == "PASS" assert evidence["scientific_json_count"] == 6 assert evidence["all_scientific_json_byte_exact"] assert not evidence["media_or_checkpoint_loaded"] modified = tmp_path / "qualification_results.json" modified.write_text(modified.read_text() + "\n") with pytest.raises(RuntimeError, match="reproduction failed"): audit.audit(tmp_path) failed = json.loads((tmp_path / "audit/statistics_reproduction_audit.json").read_text()) assert failed["status"] == "FAIL" def test_single_dominant_scene_fails_even_when_all_loso_positive(): data = fixture() for row in data[0]: if row["scene_id"] == "s0" and row["variant"] == "C1" and row["method"] == "direct": row["depth_absrel"] -= 0.2 result = run(data) checks = result["qualification_results"]["surface_checks"] assert checks["every_leave_one_scene_out_gain_positive"] assert result["static_results"]["surface_gain"]["positive_top1_contribution"] > 0.5 assert not checks["positive_top1_share_at_most_0_5"] assert result["qualification_results"]["SURFACE_TRAINING_STATUS"] == "PARTIAL"