src/mcss/mechanism_pilot/geometry_carrier_statistics.py
"""Scene-paired DEV analysis for matched RGB-input learned carriers, never model code."""
from collections import defaultdict
import numpy as np
from mcss.mechanism_pilot.direct_capacity_statistics import describe, paired, scene_values
from mcss.mechanism_pilot.observability_supervision_statistics import conditional_difference
METRICS = (
"depth_absrel",
"rgb_mse",
"rgb_ssim",
"depth_rmse",
"depth_delta1",
"opacity",
"coverage",
)
METHODS = ("direct", "anchor", "wrong_scene", "spatial_shuffle", "zero")
STATE_CHECKS = (
"shared_state_multiple_queries",
"query_after_state_seal",
"recipient_camera_preserved",
)
def dev_records(manifest):
rows = [r for r in manifest["scenes"] if r.get("split", "DEV") == "DEV"]
records = {r["scene_id"]: r for r in rows}
if not records or len(records) != len(rows):
raise ValueError("Unique nonempty DEV roster required")
forbidden = set(manifest.get("FINAL_HOLDOUT_PROHIBITED", []))
forbidden |= {
r["scene_id"]
for r in manifest["scenes"]
if r.get("split") != "DEV" and r.get("split") is not None
}
if set(records) & forbidden:
raise PermissionError("TRAIN/fresh/holdout cannot enter DEV analysis")
return records
def _macro(rows, metric, *, observation="query_id", allow_null=False):
"""First equal seeds per identical query/role, then equal queries, roles, scenes."""
groups = defaultdict(list)
for row in rows:
groups[row["scene_id"], row["role"], row[observation]].append(row[metric])
averaged, invalid = [], set()
for (scene, role, query), values in groups.items():
if any(v is None for v in values):
if not allow_null:
raise ValueError("No missing primary metric allowed")
invalid.add(scene)
continue
if not np.isfinite(values).all():
raise ValueError("Nonfinite metrics cannot be silently discarded")
averaged.append(
{"scene_id": scene, "role": role, "query_id": query, "value": float(np.mean(values))}
)
values = scene_values(averaged, "value") if averaged else {}
for scene in invalid:
values[scene] = None
return values
def _evidence(stats):
return (
stats["mean"] > 0
and stats["ci95"][0] > 0
and (stats["positive"] + stats["tied"]) / stats["n_scenes"] >= 0.75
)
def _conditional_ok(stats):
return stats["scene_coverage"] >= 0.75 and stats["ci95"] is not None and stats["ci95"][0] > 0
def validate_static(static_rows, manifest, seeds, variants, matched_rows):
"""Per-variant anchor/direct common-mask rows: complete matrix, null iff empty mask."""
records = dev_records(manifest)
expected = {
(v, seed, sid, role, fid)
for v in variants
for seed in seeds
for sid, record in records.items()
for role in ("A", "B")
for fid in record["roles"]["primary_query"]
}
actual = {
(r["variant"], r["seed"], r["scene_id"], r["role"], r["query_id"]) for r in static_rows
}
if len(actual) != len(static_rows) or actual != expected:
raise ValueError("Complete exact static anchor/direct matrix required")
for row in static_rows:
if row.get("comparison") != "anchor_minus_direct":
raise ValueError("Static rows must be anchor_minus_direct")
count, total = row["common_valid_count"], row["total_gt_valid"]
if (
type(count) is not int
or type(total) is not int
or not 0 <= count <= total
or total <= 0
):
raise ValueError("Valid static common-mask counts required")
for side in ("anchor", "direct"):
for name in ("depth_absrel", "normalized_depth_absrel"):
value = row[f"{side}_{name}"]
if (count == 0 and value is not None) or (
count > 0 and (value is None or not np.isfinite(value) or value < 0)
):
raise ValueError("Empty static mask must be null, never imputed zero")
totals = {}
for row in (*static_rows, *matched_rows):
key = (row["seed"], row["scene_id"], row["role"], row["query_id"])
totals.setdefault(key, set()).add(row["total_gt_valid"])
if any(len(values) != 1 for values in totals.values()):
raise ValueError("GT-valid denominators must agree across variants and masks")
def surface_status(gain, checks):
if gain["mean"] < 0 and gain["ci95"][1] < 0:
return "HARMFUL"
if all(checks.values()):
return "SUPPORTED"
if gain["mean"] > 0 and gain["ci95"][0] > 0:
return "PARTIAL"
return "NOT_ESTABLISHED"
def static_status(gain, checks):
if all(checks.values()):
return "SUPPORTED"
if gain["mean"] > 0 and gain["ci95"][0] > 0:
return "PARTIAL"
return "NOT_ESTABLISHED"
def validate(query, context, matched, manifest, seeds):
records = dev_records(manifest)
if not seeds or len(set(seeds)) != len(seeds):
raise ValueError("Nonempty unique frozen seeds required")
variants = {r["variant"] for r in query}
if not {"C0", "C1"} <= variants or not variants <= {"C0", "C1", "C2"}:
raise ValueError("Primary must include matched C0/C1; optional C2 remains secondary")
identities = {
(seed, sid, role, fid)
for seed in seeds
for sid, record in records.items()
for role in ("A", "B")
for fid in record["roles"]["primary_query"]
}
if any(len(set(r["roles"]["primary_query"])) < 2 for r in records.values()):
raise ValueError("At least two locked queries per scene required")
for rows, expected, id_field in (
(
query,
{(v, m, *key) for v in variants for m in METHODS for key in identities},
"query_id",
),
(
context,
{
(v, "direct", seed, sid, role, fid)
for v in variants
for seed in seeds
for sid, rec in records.items()
for role in ("A", "B")
for fid in rec["roles"]["context_a" if role == "A" else "context_b"]
},
"frame_id",
),
(matched, {(v, "direct", *key) for v in ("C0", "C1") for key in identities}, "query_id"),
):
actual = {
(r["variant"], r["method"], r["seed"], r["scene_id"], r["role"], r[id_field])
for r in rows
}
if len(actual) != len(rows) or actual != expected:
raise ValueError("Complete exact DEV/seed/variant/role/frame matrix required")
pairs = defaultdict(dict)
for row in matched:
pairs[row["seed"], row["scene_id"], row["role"], row["query_id"]][row["variant"]] = row
count, total = row["common_valid_count"], row["total_gt_valid"]
if (
type(count) is not int
or type(total) is not int
or not 0 <= count <= total
or total <= 0
):
raise ValueError("Valid common-mask counts required")
for name in ("depth_absrel", "normalized_depth_absrel"):
value = row[name]
if (count == 0 and value is not None) or (
count > 0 and (value is None or not np.isfinite(value) or value < 0)
):
raise ValueError("Empty common mask must be null, never imputed zero")
for values in pairs.values():
if any(
values["C0"][k] != values["C1"][k]
for k in ("common_valid_count", "total_gt_valid", "common_mask_hash")
):
raise ValueError("C0/C1 must use exactly the same opacity-intersection mask")
return sorted(records), sorted(variants)
def analyze(
query_rows,
context_rows,
matched_rows,
manifest,
*,
seeds,
state_use_audit,
static_matched_rows,
draws=10000,
bootstrap_seed=20260928,
):
ids, variants = validate(query_rows, context_rows, matched_rows, manifest, seeds)
validate_static(static_matched_rows, manifest, seeds, variants, matched_rows)
describe_args = dict(draws=draws, seed=bootstrap_seed)
groups = {}
for variant in variants:
for method in METHODS:
subset = [r for r in query_rows if r["variant"] == variant and r["method"] == method]
groups[f"{variant}|{method}"] = {
"variant": variant,
"method": method,
"query_metrics": {m: describe(_macro(subset, m), **describe_args) for m in METRICS},
}
cell = groups[f"{variant}|direct"]
context = [r for r in context_rows if r["variant"] == variant]
cell["context_metrics"] = {
m: describe(_macro(context, m, observation="frame_id"), **describe_args)
for m in METRICS
}
cell["generalization_gap"] = paired(
cell["query_metrics"]["depth_absrel"]["per_scene"],
cell["context_metrics"]["depth_absrel"]["per_scene"],
**describe_args,
)
def contrast(left, right):
return paired(
groups[left]["query_metrics"]["depth_absrel"]["per_scene"],
groups[right]["query_metrics"]["depth_absrel"]["per_scene"],
**describe_args,
)
gain = contrast("C0|direct", "C1|direct")
controls = {
f"{v}|{m}": contrast(f"{v}|{m}", f"{v}|direct")
for v in variants
for m in ("wrong_scene", "spatial_shuffle", "zero")
}
full = {v: contrast(f"{v}|anchor", f"{v}|direct") for v in variants}
common = {}
for metric in ("depth_absrel", "normalized_depth_absrel"):
left, right = [
_macro([r for r in matched_rows if r["variant"] == v], metric, allow_null=True)
for v in ("C0", "C1")
]
common[metric] = conditional_difference(left, right, **describe_args)
coverage = {
v: groups[f"{v}|direct"]["query_metrics"]["coverage"]["per_scene"] for v in ("C0", "C1")
}
ray_hits = {
v: _macro(
[r for r in query_rows if r["variant"] == v and r["method"] == "direct"],
"ray_hitfraction",
)
for v in ("C0", "C1")
}
if any(abs(ray_hits["C0"][s] - ray_hits["C1"][s]) > 1e-8 for s in ids):
raise ValueError("Matched variants must share frozen geometry ray-hit fractions")
opacity_checks = {
"all_scenes_coverage_at_least_0_99_ray_hitfraction": all(
coverage[v][s] >= 0.99 * ray_hits[v][s] for v in coverage for s in ids
),
"all_scenes_coverage_drop_at_most_0_01": all(
coverage["C1"][s] - coverage["C0"][s] >= -0.01 for s in ids
),
"common_mask_gain_ci_positive_with_75pct_scene_coverage": common["depth_absrel"][
"scene_coverage"
]
>= 0.75
and common["depth_absrel"]["ci95"] is not None
and common["depth_absrel"]["ci95"][0] > 0,
"opacity_normalized_gain_ci_positive_with_75pct_scene_coverage": common[
"normalized_depth_absrel"
]["scene_coverage"]
>= 0.75
and common["normalized_depth_absrel"]["ci95"] is not None
and common["normalized_depth_absrel"]["ci95"][0] > 0,
}
opacity_ok = all(opacity_checks.values())
static_common = {}
for v in variants:
subset = [r for r in static_matched_rows if r["variant"] == v]
static_common[v] = {}
for metric in ("depth_absrel", "normalized_depth_absrel"):
anchor, direct = (
_macro(
[{**r, "value": r[f"{side}_{metric}"]} for r in subset],
"value",
allow_null=True,
)
for side in ("anchor", "direct")
)
static_common[v][metric] = conditional_difference(anchor, direct, **describe_args)
static_opacity_checks = {
"C1_all_scenes_coverage_at_least_0_99_ray_hitfraction": all(
coverage["C1"][s] >= 0.99 * ray_hits["C1"][s] for s in ids
),
"C1_anchor_direct_common_mask_gain_ci_positive_with_75pct_scene_coverage": (
_conditional_ok(static_common["C1"]["depth_absrel"])
),
"C1_anchor_direct_normalized_gain_ci_positive_with_75pct_scene_coverage": (
_conditional_ok(static_common["C1"]["normalized_depth_absrel"])
),
}
static_opacity_ok = all(static_opacity_checks.values())
state_ok = state_use_audit.get("status") == "PASS" and all(
state_use_audit.get(k) is True for k in STATE_CHECKS
)
surface_checks = {
"mean_ci_and_scene_consistency": _evidence(gain),
"every_leave_one_scene_out_gain_positive": bool(gain["loso"])
and min(gain["loso"].values()) > 0,
"positive_top1_share_at_most_0_5": gain["positive_top1_contribution"] is not None
and gain["positive_top1_contribution"] <= 0.5,
"wrong_scene_ci_positive": controls["C1|wrong_scene"]["ci95"][0] > 0,
"spatial_shuffle_ci_positive": controls["C1|spatial_shuffle"]["ci95"][0] > 0,
"opacity_artifact_gate": opacity_ok,
"state_integrity": state_ok,
}
static_checks = {
"full_context_mean_ci_scene_consistency": _evidence(full["C1"]),
"wrong_scene_ci_positive": controls["C1|wrong_scene"]["ci95"][0] > 0,
"shared_sealed_state_integrity": state_ok,
"static_opacity_artifact_gate": static_opacity_ok,
}
per_seed = {}
for seed in seeds:
def seed_scene(v, method="direct", seed=seed):
return _macro(
[
r
for r in query_rows
if r["seed"] == seed and r["variant"] == v and r["method"] == method
],
"depth_absrel",
)
per_seed[str(seed)] = {
"surface_gain": paired(seed_scene("C0"), seed_scene("C1"), **describe_args),
"full_context_gain": {
v: paired(seed_scene(v, "anchor"), seed_scene(v), **describe_args) for v in variants
},
"query_absrel": {v: describe(seed_scene(v), **describe_args) for v in variants},
}
seed_robustness = (
"DIRECTION_REPLICATED"
if len(seeds) >= 2 and all(v["surface_gain"]["mean"] > 0 for v in per_seed.values())
else "NOT_ESTABLISHED"
)
qualification = {
"SURFACE_TRAINING_STATUS": surface_status(gain, surface_checks),
"STATIC_DEV_STATUS": static_status(full["C1"], static_checks),
"surface_checks": surface_checks,
"static_checks": static_checks,
"SEED_ROBUSTNESS": seed_robustness,
"FRESH_QUALIFICATION_INDEPENDENCE": manifest.get(
"fresh_status", "BLOCKED_INDEPENDENCE_UNRESOLVED"
),
"FRESH_QUALIFICATION_OPENED": False,
"FINAL_STATIC_STATUS": "NOT_ESTABLISHED",
"DYNAMIC_TTT_NEXT_STAGE_ALLOWED": False,
"DEV_does_not_establish_final_static_qualification": True,
}
return {
"static_results": {
"scene_ids": ids,
"seeds": list(seeds),
"primary": "C1_SURFACE16",
"secondary_variants": [v for v in variants if v not in ("C0", "C1")],
"groups": groups,
"surface_gain": gain,
"full_context_gain": full,
"per_seed": per_seed,
"free_space_extra_gain": contrast("C1|direct", "C2|direct")
if "C2" in variants
else None,
"aggregation": (
"equal seeds per query, then queries per role, roles per scene; scene bootstrap"
),
},
"wrong_scene_results": {
"contrasts": controls,
"bounds_caveat": (
"wrong-scene donor includes bounds; spatial shuffle preserves recipient bounds"
),
},
"state_use_results": {
"input_audit": state_use_audit,
"opacity_checks": opacity_checks,
"opacity_artifact_gate": opacity_ok,
"common_mask_gains": common,
"static_opacity_checks": static_opacity_checks,
"static_opacity_artifact_gate": static_opacity_ok,
"static_common_mask_gains": static_common,
"static_common_mask_definition": (
"per variant anchor-minus-direct on GT-valid pixels where both anchor and "
"direct opacity>1e-6; normalized uses depth/opacity.clamp_min(1e-6); empty "
"masks null; gate for STATIC_DEV_STATUS only"
),
"normalized_depth_definition": (
"rendered_depth / opacity.clamp_min(1e-6), "
"common GT-valid C0/C1 opacity>1e-6 mask; diagnostic only"
),
"primary_metric_unchanged": True,
"ray_hitfraction_per_scene": ray_hits,
"no_hit_scenes": [s for s in ids if ray_hits["C0"][s] == 0],
"coverage_gate_design": (
"Pre-training revision based on exposed historical ai_009_001 "
"all-miss bounds; no V2 outcomes. Overall retains all scenes/pixels; "
"no scene/role/bounds replacement."
),
},
"bootstrap_results": {
"draws": draws,
"seed": bootstrap_seed,
"unit": "scene, not seeds or pixels",
"surface_gain": gain,
"full_context_gain": full,
"controls": controls,
},
"qualification_results": qualification,
}
def analyze_regions(rows, manifest, *, seeds, draws=10000, bootstrap_seed=20260928):
"""Optional OBS signature; no regional result can change primary DEV gates."""
from mcss.mechanism_pilot.observability_supervision_statistics import conditional_summary
records = dev_records(manifest)
names = ("OBS0", "OBS1", "OBS2PLUS")
expected = {
(v, seed, sid, role, q, region)
for v in ("C0", "C1")
for seed in seeds
for sid, rec in records.items()
for role in ("A", "B")
for q in rec["roles"]["primary_query"]
for region in names
}
actual = {
(r["variant"], r["seed"], r["scene_id"], r["role"], r["query_id"], r["region"])
for r in rows
}
if actual != expected or len(actual) != len(rows):
raise ValueError("Complete paired OBS diagnostic rows required, including empty masks")
lookup = {
(r["variant"], r["seed"], r["scene_id"], r["role"], r["query_id"], r["region"]): r
for r in rows
}
for row in rows:
count, total, error = row["pixel_count"], row["total_valid"], row["absrel_sum"]
if (
type(count) is not int
or type(total) is not int
or not 0 <= count <= total
or total <= 0
or not np.isfinite(error)
or error < 0
):
raise ValueError(
"Finite sufficient statistics and positive overall denominator required"
)
other = lookup[
(
"C1" if row["variant"] == "C0" else "C0",
row["seed"],
row["scene_id"],
row["role"],
row["query_id"],
row["region"],
)
]
if any(row[k] != other[k] for k in ("pixel_count", "total_valid")):
raise ValueError("OBS masks cannot depend on carrier variant")
if count == 0 and (error != 0 or row.get("absrel_mean") is not None):
raise ValueError("Empty region must remain null with zero contribution")
if row["region"] == "OBS0":
parts = [
lookup[
tuple(row[k] for k in ("variant", "seed", "scene_id", "role", "query_id"))
+ (name,)
]
for name in names
]
if sum(p["pixel_count"] for p in parts) != total:
raise ValueError("OBS regions must reconstruct unfiltered overall")
options = dict(draws=draws, seed=bootstrap_seed)
output = {v: {} for v in ("C0", "C1")}
for variant in output:
for name in names:
subset = [
{
**r,
"fraction": r["pixel_count"] / r["total_valid"],
"contribution": r["absrel_sum"] / r["total_valid"],
}
for r in rows
if r["variant"] == variant and r["region"] == name
]
fraction, contribution = (_macro(subset, k) for k in ("fraction", "contribution"))
conditional = {
s: contribution[s] / fraction[s] if fraction[s] > 0 else None for s in records
}
output[variant][name] = {
"conditional_absrel": conditional_summary(conditional, **options),
"pixel_fraction": describe(fraction, **options),
"additive_error_contribution": describe(contribution, **options),
}
return {
"groups": output,
"gains": {
r: conditional_difference(
output["C0"][r]["conditional_absrel"]["per_scene"],
output["C1"][r]["conditional_absrel"]["per_scene"],
**options,
)
for r in names
},
"diagnostic_only": True,
"does_not_change_qualification": True,
}