"""Locked, stage-separated opportunity study; no validation-driven fitting.""" from __future__ import annotations import argparse import hashlib import itertools import json import platform import subprocess import sys from datetime import UTC, datetime from pathlib import Path import numpy as np import torch from PIL import Image from mcss.vision_probe.adaptation import ( ACTIONS, FastState, Readout, fit_readout, masks, materialize, proposals, ) from mcss.vision_probe.experiment import _cached_extract, _module_hash from mcss.vision_probe.features import FrozenDinoExtractor ROOT = Path(__file__).resolve().parents[3] REPORT = ROOT / "docs/experiments/EXP-2D-20260926-opportunity-selector-v1" WORK = ROOT / "outputs/EXP-2D-20260926-opportunity-selector-v1" DATA = ROOT / "data/vision_2d_probe/davis2017_trainval_480p/DAVIS" WEIGHTS = ROOT / "data/vision_2d_probe/weights/dinov2_vits14_pretrain.pth" TRAJECTORIES = list(itertools.product(ACTIONS, repeat=2)) CONFIG = { "schema": "opportunity-selector-v1", "resolutions": [224, 448], "primary_resolution": 448, "primary_metric": "J", "seed": 20260926, "bootstrap_seed": 20260926, "bootstrap_samples": 10000, "eta": 5.0, "eta_rule": "fixed from historical discovery; no new eta search", "latent_width": 64, "rgb_ridge": 0.1, "max_increment_norm": 0.15, "boundary_tolerance_diagonal": 0.008, "probability_threshold": 0.5, "size_bucket_thresholds": [0.005, 0.02, 0.1], "ridge_alphas": [0.1, 1.0, 10.0, 100.0], "abstention_thresholds": [0.0, 0.001, 0.005, 0.01, 0.02], "rgb_thresholds": [0.0, 0.01, 0.05, 0.1, 0.2, 0.5, 1.0], "rank_relative_tolerance": 1e-5, "rank_absolute_tolerance": 1e-8, "opportunity_strong": { "minimum_positive_sequences": 4, "top1_max": 0.5, "sequence_ci_lower_gt": 0, "loso_min_gt": 0, "non_tiny_gain_gt": 0, }, "opportunity_moderate": {"minimum_positive_sequences": 2, "top1_max_exclusive": 0.8}, "identifiability_strong": {"both_ci_lower_gt": 0, "harmful_among_writes_max": 0.25}, "reserve_policy": "remains sealed; no confirmation in this v1", "validation_history": "same 16 historical internal validation sequences; not a fresh holdout", "aggregation": "objects within pair, both target pairs within sequence, equal sequence mean", "tie_rule": "OFF,A,B,ALL Cartesian order, earliest exact maximum", "correspondence": "cosine hard nearest source token, no target masks", "pixel_readout": "source object area occupancy, nearest transfer, bilinear upsample, >=0.5", "F_scope": "approximate binary boundary F, secondary; not official DAVIS J&F", "stages": [ "fit", "discovery_opportunity", "discovery_selectors", "lock", "validation", "report", ], } def sha(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() def write_json(path: Path, value: object) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(value, indent=2, sort_keys=True, allow_nan=False) + "\n") def read_json(path: Path): return json.loads(path.read_text()) def source_hashes() -> dict[str, str]: files = [ p for p in (ROOT / "src/mcss/vision_probe").glob("*.py") if p.name != "opportunity_reporting.py" ] files += list((ROOT / "third_party/dinov2").rglob("*.py")) files += [ ROOT / "scripts/run_opportunity_probe.py", ROOT / "scripts/audit_opportunity_probe.py", ROOT / "pyproject.toml", ROOT / "uv.lock", ] return {str(p.relative_to(ROOT)): sha(p) for p in sorted(files) if p.exists()} def validate_splits(splits: dict) -> None: groups = [ set(splits[k]) for k in ("fit", "discovery", "validation", "reserve", "official_val_sealed") ] if any( len(g) != len(splits[k]) for g, k in zip( groups, ("fit", "discovery", "validation", "reserve", "official_val_sealed"), strict=True, ) ): raise ValueError("Duplicate split identity") if any(a & b for a, b in itertools.combinations(groups, 2)): raise ValueError("Split overlap") if list(map(len, groups)) != [24, 12, 16, 8, 30]: raise ValueError("Frozen split counts changed") class MediaGuard: """Fail closed for any image/mask open or directory listing outside allowed identities.""" def __init__(self, data: Path, splits: dict): self.data = data.resolve() self.allowed: set[str] = set() self.events: list[dict] = [] self.splits = splits def allow(self, split: str) -> None: if split not in ("fit", "discovery", "validation"): raise ValueError("Sealed split cannot be enabled") self.allowed = set(self.splits[split]) self.events.append({"event": "allow", "split": split, "utc": datetime.now(UTC).isoformat()}) def check_path(self, path: str | Path) -> None: try: rel = Path(path).resolve().relative_to(self.data) except (ValueError, TypeError): return parts = rel.parts if parts and parts[0] in ("JPEGImages", "Annotations"): if len(parts) < 3 or parts[2] not in self.allowed: raise PermissionError(f"Sealed/unauthorized media access: {rel}") def hook(self, event: str, args: tuple) -> None: if ( event in ("open", "os.listdir", "os.scandir") and args and isinstance(args[0], (str, bytes, Path)) ): path = args[0].decode() if isinstance(args[0], bytes) else args[0] self.check_path(path) try: rel = Path(path).resolve().relative_to(self.data) except ValueError: return if rel.parts and rel.parts[0] in ("JPEGImages", "Annotations"): self.events.append({"event": event, "path": str(rel)}) def frames_for(name: str) -> list[Path]: frames = sorted((DATA / "JPEGImages/480p" / name).glob("*.jpg")) if len(frames) < 3: raise ValueError(f"Insufficient frames: {name}") return [frames[0], frames[len(frames) // 3], frames[2 * len(frames) // 3]] def prepare() -> None: if REPORT.exists(): raise FileExistsError("Refusing to overwrite experiment directory") # Use original explicit identity protocol, not a scan of sealed media. splits = read_json(ROOT / "docs/experiments/EXP-2D-20260921-corrected/protocol.json")["splits"] validate_splits(splits) train = set((DATA / "ImageSets/2017/train.txt").read_text().split()) if train != set().union( *(set(splits[k]) for k in ("fit", "discovery", "validation", "reserve")) ): raise ValueError("Official train identities changed") REPORT.mkdir(parents=True) WORK.mkdir(parents=True, exist_ok=True) write_json(REPORT / "config.json", CONFIG) write_json(REPORT / "split_manifest.json", splits) env = { "python": sys.version, "platform": platform.platform(), "torch": torch.__version__, "cuda": torch.version.cuda, "gpu": torch.cuda.get_device_name(0), "weight_sha256": sha(WEIGHTS), "created_utc": datetime.now(UTC).isoformat(), "submodules": subprocess.check_output(["git", "submodule", "status"], cwd=ROOT, text=True), "packages": subprocess.check_output( [str(Path.home() / ".local/bin/uv"), "pip", "freeze", "--python", sys.executable], cwd=ROOT, text=True, ), } write_json(REPORT / "environment.json", env) (REPORT / "git_commit.txt").write_text( subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=ROOT, text=True) + "Working changes are fingerprinted in validation_lock.json; base commit is n" "ot experiment code.\n" ) (REPORT / "STATUS.md").write_text("PHASE=PREREGISTERED\nFINAL_INTEGRITY=PENDING\n") (REPORT / "README.md").write_text( "# Opportunity + visible selector v1\n" "\n" "Pre-registered before new validation evaluation. Primary: pixel-space objec" "t J at448;224 control. Preserve 64-dimensional latent and all16 two-step tr" "ajectories. No reserve or official validation access.\n" "\n" "Protocol details are in config.json. Equal sequence aggregation; 10,000 seq" "uence bootstrap draws. Historical internal validation identities are reused" ", so this is not a fresh final holdout. No parameter changes after validati" "on starts.\n" ) (REPORT / "commands.sh").write_text( "#!/usr/bin/env bash\n" "set -euo pipefail\n" "cd /home/zonghan/measurement-scene-state\n" "export OMP_NUM_THREADS=4\n" "PY=.venv/bin/python\n" "$PY scripts/run_opportunity_probe.py prepare\n" "$PY scripts/run_opportunity_probe.py discovery\n" "$PY scripts/run_opportunity_probe.py lock\n" "$PY scripts/run_opportunity_probe.py validation\n" "$PY scripts/run_opportunity_probe.py report\n" ) write_json( REPORT / "preregistration.json", { "utc": datetime.now(UTC).isoformat(), "config_sha256": sha(REPORT / "config.json"), "split_sha256": sha(REPORT / "split_manifest.json"), "source_hashes_at_registration": source_hashes(), }, ) def rank_record(delta: torch.Tensor, decoder: torch.Tensor, step: int, module: str) -> dict: matrix = delta.double() singular = torch.linalg.svdvals(matrix) tol = max(1e-8, float(singular[0]) * 1e-5) basis = torch.linalg.svd(decoder.double(), full_matrices=False).U residual = matrix - basis @ (basis.T @ matrix) return { "step": step, "module": module, "singular_values": singular.tolist(), "numerical_rank": int((singular > tol).sum()), "tolerance": tol, "span_residual": float(residual.norm() / matrix.norm().clamp_min(1e-15)), "norm": float(matrix.norm()), "span": "column span of frozen RGB decoder (64x3)", } def candidate_states(target, readout: Readout, eta: float): z = readout.project(target.features) result = [] for trajectory in TRAJECTORIES: state = FastState.zero(64) audits = [] for step, action in enumerate(trajectory): support, _ = masks(*z.shape[:2], step) old = state state = proposals(z, target.rgb, readout, old, support, eta)[action] for module in ("a", "b"): audits.append( rank_record( getattr(state, module) - getattr(old, module), readout.decoder, step, module.upper(), ) ) adapted = readout.restore(target.features, z, materialize(z, state)) result.append((state, adapted, audits)) return z, result def extraction(size: int, split: str, splits: dict, guard: MediaGuard): guard.allow(split) extractor = FrozenDinoExtractor(ROOT / "third_party/dinov2", WEIGHTS, "cuda", size) features = {} for name in splits[split]: paths = frames_for(name) features[name] = [ ( path, _cached_extract( extractor, path, WORK / f"cache/{size}", sha(WEIGHTS), _module_hash(), size ), ) for path in paths ] print(f"extract resolution={size} split={split} sequence={name}", flush=True) del extractor torch.cuda.empty_cache() return features def fit_resolution(size: int, splits: dict, guard: MediaGuard) -> Readout: features = extraction(size, "fit", splits, guard) items = [item for frames in features.values() for _, item in frames] readout = fit_readout( torch.cat([i.features for i in items]), torch.cat([i.rgb for i in items]), 64, 0.1 ) torch.save(vars(readout), WORK / f"readout_{size}.pt") write_json( REPORT / f"readout_{size}_provenance.json", { "fit_sequences": splits["fit"], "mask_access": False, "resolution": size, "latent_width": 64, "sha256": sha(WORK / f"readout_{size}.pt"), }, ) return readout def run_pairs( size: int, split: str, splits: dict, guard: MediaGuard, readout: Readout, artifact=None ): from mcss.vision_probe.opportunity_metrics import evaluate_pair from mcss.vision_probe.opportunity_selectors import ( enrich_candidates, select_candidates, visible_features, ) features = extraction(size, split, splits, guard) records = [] for name, frames in features.items(): source_path, source = frames[0] for slot, (target_path, target) in enumerate(frames[1:], 1): z, candidates = candidate_states(target, readout, CONFIG["eta"]) visible = [ visible_features( source.features, target.features, adapted, target.rgb, readout, state, z ) for state, adapted, _ in candidates ] visible = enrich_candidates(visible, [a for _, a, _ in candidates]) choices = ( {} if artifact is None else select_candidates(visible, [list(t) for t in TRAJECTORIES], artifact) ) # Commit deployable choices BEFORE opening any target GT. with (REPORT / f"{split}_decisions.jsonl").open("a") as handle: handle.write( json.dumps( { "resolution": size, "sequence": name, "target_slot": slot, "choices": choices, "before_target_gt": True, } ) + "\n" ) source_mask_path = DATA / "Annotations/480p" / name / (source_path.stem + ".png") target_mask_path = DATA / "Annotations/480p" / name / (target_path.stem + ".png") with Image.open(source_mask_path) as im: source_mask = np.asarray(im).copy() with Image.open(target_mask_path) as im: target_mask = np.asarray(im).copy() for index, (trajectory, (_, adapted, audit), visible_row) in enumerate( zip(TRAJECTORIES, candidates, visible, strict=True) ): metric = evaluate_pair(source.features, adapted, source_mask, target_mask) row = { "resolution": size, "split": split, "sequence": name, "target_slot": slot, "source_frame": source_path.name, "source_image_sha256": source.input_sha256, "target_image_sha256": target.input_sha256, "source_mask_sha256": sha(source_mask_path), "target_mask_sha256": sha(target_mask_path), "target_frame": target_path.name, "trajectory": list(trajectory), "visible": visible_row, "rank_audit": audit, "selected_methods": [m for m, i in choices.items() if i == index], **metric, } records.append(row) print( f"evaluate resolution={size} split={split} sequence={name} slot={slot}", flush=True ) return records def save_rows(path: Path, rows: list[dict]) -> None: path.write_text( "".join(json.dumps(row, sort_keys=True, allow_nan=False) + "\n" for row in rows) ) def load_rows(path: Path) -> list[dict]: return [json.loads(line) for line in path.read_text().splitlines() if line.strip()] def discovery(splits: dict, guard: MediaGuard): if (REPORT / "discovery_raw.jsonl").exists(): raise FileExistsError("Discovery already completed; do not overwrite") rows = [] for size in CONFIG["resolutions"]: readout = fit_resolution(size, splits, guard) rows.extend(run_pairs(size, "discovery", splits, guard, readout)) save_rows(WORK / "discovery_progress.jsonl", rows) save_rows(REPORT / "discovery_raw.jsonl", rows) from mcss.vision_probe.opportunity_analysis import opportunity_summary write_json(REPORT / "discovery_opportunity.json", opportunity_summary(rows)) (REPORT / "STATUS.md").write_text( "PHASE=DISCOVERY_OPPORTUNITY_COMPLETE\nFINAL_INTEGRITY=PENDING\n" ) def lock_selectors(splits: dict): from mcss.vision_probe.opportunity_selectors import fit_selectors, select_candidates if (REPORT / "validation_lock.json").exists(): raise FileExistsError("Selectors already locked") if not (REPORT / "discovery_opportunity.json").exists(): raise RuntimeError("Opportunity analysis must precede selector fitting") rows = load_rows(REPORT / "discovery_raw.jsonl") artifact = fit_selectors(rows, CONFIG) write_json(REPORT / "selectors.json", artifact) for size in CONFIG["resolutions"]: for name in splits["discovery"]: for slot in (1, 2): pair = [ r for r in rows if r["resolution"] == size and r["sequence"] == name and r["target_slot"] == slot ] choices = select_candidates( [r["visible"] for r in pair], [r["trajectory"] for r in pair], artifact["by_resolution"][str(size)], ) for index, row in enumerate(pair): row["selected_methods"] = [m for m, i in choices.items() if i == index] save_rows(REPORT / "raw_results.jsonl", rows) lock = { "utc": datetime.now(UTC).isoformat(), "source_hashes": source_hashes(), "config_sha256": sha(REPORT / "config.json"), "split_sha256": sha(REPORT / "split_manifest.json"), "selectors_sha256": sha(REPORT / "selectors.json"), "weights_sha256": sha(WEIGHTS), "discovery_sha256": sha(REPORT / "discovery_raw.jsonl"), "readout_sha256": {str(s): sha(WORK / f"readout_{s}.pt") for s in CONFIG["resolutions"]}, "fit_split": "discovery", "validation_started": False, } write_json(REPORT / "validation_lock.json", lock) (REPORT / "STATUS.md").write_text("PHASE=LOCKED_BEFORE_VALIDATION\nFINAL_INTEGRITY=PENDING\n") def verify_lock() -> dict: lock = read_json(REPORT / "validation_lock.json") checks = { "source_hashes": source_hashes(), "config_sha256": sha(REPORT / "config.json"), "split_sha256": sha(REPORT / "split_manifest.json"), "selectors_sha256": sha(REPORT / "selectors.json"), "weights_sha256": sha(WEIGHTS), "discovery_sha256": sha(REPORT / "discovery_raw.jsonl"), "readout_sha256": {str(s): sha(WORK / f"readout_{s}.pt") for s in CONFIG["resolutions"]}, } for key, value in checks.items(): if lock[key] != value: raise RuntimeError(f"Validation lock mismatch: {key}") if lock["fit_split"] != "discovery": raise ValueError("Selectors not fitted on discovery") return lock def validation(splits: dict, guard: MediaGuard): verify_lock() marker = REPORT / "validation_started.json" if marker.exists(): raise RuntimeError("Validation already started; explicit bug audit required for any rerun") write_json( marker, {"utc": datetime.now(UTC).isoformat(), "lock_sha256": sha(REPORT / "validation_lock.json")}, ) artifact = read_json(REPORT / "selectors.json") rows = load_rows(REPORT / "raw_results.jsonl") for size in CONFIG["resolutions"]: readout = Readout(**torch.load(WORK / f"readout_{size}.pt", weights_only=True)) new = run_pairs( size, "validation", splits, guard, readout, artifact["by_resolution"][str(size)] ) rows.extend(new) save_rows(REPORT / "raw_results.jsonl", rows) verify_lock() write_json( REPORT / "validation_complete.json", { "utc": datetime.now(UTC).isoformat(), "raw_sha256": sha(REPORT / "raw_results.jsonl"), "n_rows": len(rows), }, ) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("stage", choices=["prepare", "discovery", "lock", "validation", "report"]) args = parser.parse_args() torch.set_num_threads(4) torch.manual_seed(CONFIG["seed"]) np.random.seed(CONFIG["seed"]) torch.backends.cuda.matmul.allow_tf32 = False torch.backends.cudnn.allow_tf32 = False if args.stage == "prepare": prepare() return if read_json(REPORT / "config.json") != CONFIG: raise RuntimeError("Runtime config differs from preregistration") splits = read_json(REPORT / "split_manifest.json") validate_splits(splits) guard = MediaGuard(DATA, splits) sys.addaudithook(guard.hook) try: if args.stage == "discovery": discovery(splits, guard) elif args.stage == "lock": lock_selectors(splits) elif args.stage == "validation": validation(splits, guard) else: verify_lock() from mcss.vision_probe.opportunity_analysis import analyze analyze( REPORT / "raw_results.jsonl", REPORT, CONFIG, read_json(REPORT / "selectors.json") ) finally: write_json(REPORT / f"access_audit_{args.stage}.json", guard.events) if __name__ == "__main__": main()