"""Generate the canonical M7-02 numerical benchmark matrix from public evidence.""" from __future__ import annotations import argparse import hashlib import json from pathlib import Path from typing import Any, cast from comfyui_sigmax.core import SigmaDomain, numerical_fingerprint ROOT = Path(__file__).resolve().parents[1] TARGET = ROOT / "comfyui_sigmax" / "benchmarks" / "numerical_matrix_v1.json" SOURCE_PATHS = ( "comfyui_sigmax/workflows/fixtures.json", "tests/benchmarks/fixtures/known_good_host_attempts_v1.json", "tests/conformance/fixtures/capability_receipt_conformance_v1.json", "tests/parity/fixtures/krea2_native_euler_parity_v1.json", "tests/parity/fixtures/krea2_raw_parity_v1.json", "tests/parity/fixtures/krea2_turbo_parity_v1.json", "tests/golden/aura_flow_v0_2.json", "tests/golden/anima_v1.json", "tests/golden/wan_v1.json", "tests/golden/ltx_v1.json", ) def _read_json(path: str) -> dict[str, Any]: value = json.loads((ROOT / path).read_text(encoding="utf-8")) if not isinstance(value, dict): raise RuntimeError(f"{path} must contain a JSON object") return cast(dict[str, Any], value) def _canonical(value: object) -> bytes: return json.dumps( value, ensure_ascii=False, allow_nan=False, separators=(",", ":"), sort_keys=True, ).encode("utf-8") def _identity(payload: bytes) -> str: return "sha256:" + hashlib.sha256(payload).hexdigest() def _source_records(sources: dict[str, dict[str, Any]]) -> list[dict[str, object]]: return [ { "path": path, "schema": sources[path]["schema"], "sha256": _identity((ROOT / path).read_bytes()), "status": sources[path].get("status"), } for path in SOURCE_PATHS ] def _runtime( *, device: str, dtype: str, python: str | None = None, torch: str | None = None, numpy: str | None = None, diffusers: str | None = None, comfyui: str | None = None, comfyui_revision_chunks: list[str] | None = None, sigmax_revision_chunks: list[str] | None = None, ) -> dict[str, object]: return { "comfyui": comfyui, "comfyui_revision_chunks": comfyui_revision_chunks, "device": device, "diffusers": diffusers, "dtype": dtype, "numpy": numpy, "python": python, "sigmax_revision_chunks": sigmax_revision_chunks, "torch": torch, } def _counts( *, requested_transitions: int, effective_transitions: int, requested_model_evaluations: int, effective_model_evaluations: int, ) -> dict[str, int]: return { "effective_model_evaluations": effective_model_evaluations, "effective_transitions": effective_transitions, "requested_model_evaluations": requested_model_evaluations, "requested_transitions": requested_transitions, } def _workload( *, requested_width: int, requested_height: int, effective_width: int, effective_height: int, transitions: int, ) -> dict[str, object]: return { "effective": { "height": effective_height, "transitions": transitions, "width": effective_width, }, "requested": { "height": requested_height, "transitions": transitions, "width": requested_width, }, } def _baseline_summary(comparisons: dict[str, Any]) -> dict[str, object]: result: dict[str, object] = {} for name in ("diffusers_float32", "krea_float64"): comparison = comparisons[name] result[name] = { "device": comparison["device"], "dtype": comparison["dtype"], "fingerprint": comparison["fingerprint"], "max_abs_error": comparison["max_abs_error"], "mean_abs_error": comparison["mean_abs_error"], "status": comparison["status"], "tolerance": comparison["tolerance"], } return result def _profile( *, identifier: str, variant: str, evidence: str, recipe: str, ) -> dict[str, str]: return { "evidence": evidence, "id": identifier, "recipe": recipe, "variant": variant, "version": "1", } def _parity_rows( turbo: dict[str, Any], raw: dict[str, Any], ) -> list[dict[str, object]]: rows: list[dict[str, object]] = [] turbo_runtime = turbo["environment"] for case in turbo["cases"]: steps = cast(int, case["steps"]) comparisons = cast(dict[str, Any], case["comparisons"]) rows.append( { "baselines": _baseline_summary(comparisons), "capability": {"level": "allow", "reasons": ["compatible"]}, "determinism": None, "evidence": { "artifact": { "construction_fingerprint": None, "numerical_fingerprint": comparisons["krea_float64"]["fingerprint"], }, "receipt_fingerprint": None, "source_ids": ["parity.turbo"], }, "execution": { "counts": _counts( requested_transitions=steps, effective_transitions=0, requested_model_evaluations=0, effective_model_evaluations=0, ), "rng_ownership": { "model": "none", "sampler": "none", "schedule": "none", }, "status": "not_executed", }, "id": f"parity.turbo.{steps}", "lane": "turbo_schedule_parity", "model_weights_present": False, "profile": _profile( identifier=turbo["profile"]["id"], variant="Turbo", evidence=case["evidence"], recipe=f"krea2.turbo.official-{steps}", ), "repeat": None, "runtime": _runtime( device=turbo_runtime["device"], dtype="float64+float32", torch=turbo_runtime["torch"], numpy=turbo_runtime["numpy"], diffusers=turbo_runtime["diffusers"], ), "schedule": {"image_seq_len": None, "mu": "1.15"}, "weight_variant": "none", "workload": _workload( requested_width=1024, requested_height=1024, effective_width=1024, effective_height=1024, transitions=steps, ), } ) raw_runtime = raw["environment"] for case in raw["cases"]: comparisons = cast(dict[str, Any], case["comparisons"]) rows.append( { "baselines": _baseline_summary(comparisons), "capability": {"level": "allow", "reasons": ["compatible"]}, "determinism": None, "evidence": { "artifact": { "construction_fingerprint": None, "numerical_fingerprint": comparisons["krea_float64"]["fingerprint"], }, "receipt_fingerprint": None, "source_ids": ["parity.raw"], }, "execution": { "counts": _counts( requested_transitions=case["steps"], effective_transitions=0, requested_model_evaluations=0, effective_model_evaluations=0, ), "rng_ownership": { "model": "none", "sampler": "none", "schedule": "none", }, "status": "not_executed", }, "id": f"parity.raw.{case['case_id']}", "lane": "raw_schedule_parity", "model_weights_present": False, "profile": _profile( identifier=raw["profile"]["id"], variant="RAW", evidence=case["evidence"], recipe=case["recipe_id"], ), "repeat": None, "runtime": _runtime( device=raw_runtime["device"], dtype="float64+float32", torch=raw_runtime["torch"], numpy=raw_runtime["numpy"], diffusers=raw_runtime["diffusers"], ), "schedule": { "image_seq_len": case["image_seq_len"], "mu": case["mu"], }, "weight_variant": "none", "workload": _workload( requested_width=case["requested_width"], requested_height=case["requested_height"], effective_width=case["effective_width"], effective_height=case["effective_height"], transitions=case["steps"], ), } ) return rows def _anima_rows(anima: dict[str, Any]) -> list[dict[str, object]]: """Project Anima's independent schedule goldens into the benchmark contract. These are schedule-only parity rows. They deliberately carry no host receipt, model execution count, or image-quality assertion; Base/Aesthetic's compact four-step vectors are marked ``modified`` because they are outside their released 30-50 step recipe range. """ variant_names = { "base": ("Base", "anima.base.framework-reference"), "aesthetic": ("Aesthetic", "anima.aesthetic.framework-reference"), "turbo": ("Turbo", "anima.turbo.framework-reference"), } rows: list[dict[str, object]] = [] for case in cast(list[dict[str, Any]], anima["cases"]): variant_key = cast(str, case["variant"]) try: variant, profile_id = variant_names[variant_key] except KeyError as error: raise RuntimeError(f"unsupported Anima golden variant: {variant_key}") from error steps = cast(int, case["steps"]) float64 = tuple(cast(float, value) for value in cast(list[object], case["float64"])) float32 = tuple(cast(float, value) for value in cast(list[object], case["float32"])) float64_fingerprint = numerical_fingerprint( float64, domain=SigmaDomain.UNIT_FLOW, precision="float64", ) float32_fingerprint = numerical_fingerprint( float32, domain=SigmaDomain.UNIT_FLOW, precision="float32", ) evidence = "framework_reference" if variant_key == "turbo" else "modified" rows.append( { "baselines": { "anima_float32": { "device": "cpu", "dtype": "float32", "fingerprint": float32_fingerprint, "max_abs_error": "0", "mean_abs_error": "0", "status": "PASS", "tolerance": "0", }, "anima_float64": { "device": "cpu", "dtype": "float64", "fingerprint": float64_fingerprint, "max_abs_error": "0", "mean_abs_error": "0", "status": "PASS", "tolerance": "0", }, }, "capability": {"level": "allow", "reasons": ["compatible"]}, "determinism": None, "evidence": { "artifact": { "construction_fingerprint": None, "numerical_fingerprint": float64_fingerprint, }, "receipt_fingerprint": None, "source_ids": ["parity.anima"], }, "execution": { "counts": _counts( requested_transitions=steps, effective_transitions=0, requested_model_evaluations=0, effective_model_evaluations=0, ), "rng_ownership": { "model": "none", "sampler": "none", "schedule": "none", }, "status": "not_executed", }, "id": f"parity.anima.{variant_key}-{steps}", "lane": "anima_schedule_parity", "model_weights_present": False, "profile": _profile( identifier=profile_id, variant=variant, evidence=evidence, recipe=f"{profile_id}-{steps}", ), "repeat": None, "runtime": _runtime(device="cpu", dtype="float64+float32"), "schedule": {"image_seq_len": None, "mu": "3.0"}, "weight_variant": "none", "workload": _workload( requested_width=1024, requested_height=1024, effective_width=1024, effective_height=1024, transitions=steps, ), } ) return rows def _wan_rows(wan: dict[str, Any]) -> list[dict[str, object]]: """Project Wan's independent vectors into schedule-only benchmark rows.""" rows: list[dict[str, object]] = [] for case in cast(list[dict[str, Any]], wan["cases"]): profile_id = cast(str, case["profile"]) steps = cast(int, case["steps"]) resolution = cast(str, case["resolution"]) ratio = cast(float, case["ratio"]) float64 = tuple(cast(float, value) for value in cast(list[object], case["float64"])) float32 = tuple(cast(float, value) for value in cast(list[object], case["float32"])) float64_fingerprint = numerical_fingerprint( float64, domain=SigmaDomain.UNIT_FLOW, precision="float64", ) float32_fingerprint = numerical_fingerprint( float32, domain=SigmaDomain.UNIT_FLOW, precision="float32", ) evidence = "official" if ".official-native" in profile_id else "framework_reference" if resolution == "480p": width, height = 832, 480 elif resolution == "720p": width, height = 1280, 720 else: width, height = 1024, 1024 rows.append( { "baselines": { "wan_float32": { "device": "cpu", "dtype": "float32", "fingerprint": float32_fingerprint, "max_abs_error": "0", "mean_abs_error": "0", "status": "PASS", "tolerance": "0", }, "wan_float64": { "device": "cpu", "dtype": "float64", "fingerprint": float64_fingerprint, "max_abs_error": "0", "mean_abs_error": "0", "status": "PASS", "tolerance": "0", }, }, "capability": {"level": "allow", "reasons": ["compatible"]}, "determinism": None, "evidence": { "artifact": { "construction_fingerprint": None, "numerical_fingerprint": float64_fingerprint, }, "receipt_fingerprint": None, "source_ids": ["parity.wan"], }, "execution": { "counts": _counts( requested_transitions=steps, effective_transitions=0, requested_model_evaluations=0, effective_model_evaluations=0, ), "rng_ownership": {"model": "none", "sampler": "none", "schedule": "none"}, "status": "not_executed", }, "id": f"parity.wan.{profile_id}-{steps}", "lane": "wan_schedule_parity", "model_weights_present": False, "profile": _profile( identifier=profile_id, variant="Wan", evidence=evidence, recipe=f"{profile_id}-{steps}", ), "repeat": None, "runtime": _runtime(device="cpu", dtype="float64+float32"), "schedule": {"image_seq_len": None, "mu": str(ratio)}, "weight_variant": "none", "workload": _workload( requested_width=width, requested_height=height, effective_width=width, effective_height=height, transitions=steps, ), } ) return rows def _ltx_rows(ltx: dict[str, Any]) -> list[dict[str, object]]: """Project adaptive LTX vectors into schedule-only benchmark rows. Distilled LTX vectors remain golden-only because the benchmark schema's ``mu`` field represents an executed adaptive shift; recording a fabricated zero shift would hide that these presets do not use token-derived shifting. """ variant_by_profile = { "ltxv.0.9.8.dev": "LTXV 0.9.8", "ltx2.19b.dev": "LTX-2 19B", "ltx2.3.22b.dev": "LTX-2.3 22B", } shift_by_tokens = {1024: "0.95", 4096: "2.05"} rows: list[dict[str, object]] = [] for case in cast(list[dict[str, Any]], ltx["cases"]): profile_id = cast(str, case["profile"]) token_count = case["token_count"] if profile_id not in variant_by_profile or not isinstance(token_count, int): continue shift = shift_by_tokens.get(token_count) if shift is None: raise RuntimeError(f"LTX benchmark token anchor is not pinned: {token_count}") steps = cast(int, case["steps"]) float64 = tuple(cast(float, value) for value in cast(list[object], case["float64"])) float32 = tuple(cast(float, value) for value in cast(list[object], case["float32"])) float64_fingerprint = numerical_fingerprint( float64, domain=SigmaDomain.UNIT_FLOW, precision="float64", ) float32_fingerprint = numerical_fingerprint( float32, domain=SigmaDomain.UNIT_FLOW, precision="float32", ) rows.append( { "baselines": { "ltx_float32": { "device": "cpu", "dtype": "float32", "fingerprint": float32_fingerprint, "max_abs_error": "0", "mean_abs_error": "0", "status": "PASS", "tolerance": "0", }, "ltx_float64": { "device": "cpu", "dtype": "float64", "fingerprint": float64_fingerprint, "max_abs_error": "0", "mean_abs_error": "0", "status": "PASS", "tolerance": "0", }, }, "capability": {"level": "allow", "reasons": ["compatible"]}, "determinism": None, "evidence": { "artifact": { "construction_fingerprint": None, "numerical_fingerprint": float64_fingerprint, }, "receipt_fingerprint": None, "source_ids": ["parity.ltx"], }, "execution": { "counts": _counts( requested_transitions=steps, effective_transitions=0, requested_model_evaluations=0, effective_model_evaluations=0, ), "rng_ownership": {"model": "none", "sampler": "none", "schedule": "none"}, "status": "not_executed", }, "id": f"parity.ltx.{profile_id}-{steps}", "lane": "ltx_schedule_parity", "model_weights_present": False, "profile": _profile( identifier=profile_id, variant=variant_by_profile[profile_id], evidence="official", recipe=f"{profile_id}-{steps}", ), "repeat": None, "runtime": _runtime(device="cpu", dtype="float64+float32"), "schedule": {"image_seq_len": None, "mu": shift}, "weight_variant": "none", "workload": _workload( requested_width=1024, requested_height=1024, effective_width=1024, effective_height=1024, transitions=steps, ), } ) return rows def _repeat(lane: dict[str, Any]) -> dict[str, object]: return { "accepted": lane["accepted"], "first_status": lane["first_status"], "repeat_status": lane["repeat_status"], "stable": lane["accepted"] and lane["first_status"] == lane["repeat_status"], "transition": lane["transition"], } def _workflow_rows( workflows: dict[str, Any], host_attempts: dict[str, Any], raw_parity: dict[str, Any], ) -> list[dict[str, object]]: host = host_attempts["host"] lanes = {lane["id"]: lane for lane in host_attempts["lanes"]} raw_cases = cast(list[dict[str, Any]], raw_parity["cases"]) rows: list[dict[str, object]] = [] for fixture in workflows["fixtures"]: lane = lanes.get(f"h2.{fixture['id']}") # A packaged workflow enters the verified matrix only after a real-host attempt exists. if lane is None: continue scheduler = next( ( node for node in fixture["workflow"]["nodes"] if node["type"] == "Sigmax.Krea2SigmaScheduler" ), None, ) # Anima has a separate golden/parity lane; a non-Krea fixture must not # be forced through the Krea widget schema. if scheduler is None: continue variant, steps, width, height, strict_official, _, _ = scheduler["widgets_values"] effective_width = (width + 15) // 16 * 16 effective_height = (height + 15) // 16 * 16 metadata = fixture["workflow"]["extra"]["comfyui_sigmax"]["metadata"] artifact = metadata["artifact"] receipt = metadata["receipts"][0] if variant == "RAW": parity_case = next( case for case in raw_cases if case["steps"] == steps and case["effective_width"] == effective_width and case["effective_height"] == effective_height ) evidence = "official" if strict_official else "framework_reference" recipe = parity_case["recipe_id"] schedule = { "image_seq_len": parity_case["image_seq_len"], "mu": parity_case["mu"], } else: evidence = "official" recipe = "krea2.turbo.official-8" schedule = {"image_seq_len": None, "mu": "1.15"} rows.append( { "baselines": None, "capability": {"level": "allow", "reasons": ["compatible"]}, "determinism": None, "evidence": { "artifact": { "construction_fingerprint": artifact["construction_fingerprint"], "numerical_fingerprint": artifact["numerical_fingerprint"], }, "receipt_fingerprint": receipt["receipt_fingerprint"], "source_ids": ["workflow.fixtures", "host.known_good"], }, "execution": { "counts": _counts( requested_transitions=steps, effective_transitions=0, requested_model_evaluations=steps, effective_model_evaluations=0, ), "rng_ownership": { "model": "none", "sampler": "none", "schedule": "none", }, "status": "not_executed", }, "id": f"host.h2.{fixture['id']}", "lane": "h2_workflow", "model_weights_present": False, "profile": _profile( identifier=metadata["profile"]["id"], variant=variant, evidence=evidence, recipe=recipe, ), "repeat": _repeat(lane), "runtime": _runtime( device="cpu", dtype="float64", comfyui=host["version"], comfyui_revision_chunks=host["revision_chunks"], sigmax_revision_chunks=host_attempts["sigmax_revision_chunks"], ), "schedule": schedule, "weight_variant": "none", "workload": _workload( requested_width=width, requested_height=height, effective_width=effective_width, effective_height=effective_height, transitions=steps, ), } ) return rows def _native_euler_row( native: dict[str, Any], workflows: dict[str, Any], host_attempts: dict[str, Any], ) -> dict[str, object]: fixture = next( fixture for fixture in workflows["fixtures"] if fixture["id"] == "krea2-turbo-1024" ) metadata = fixture["workflow"]["extra"]["comfyui_sigmax"]["metadata"] lane = next(lane for lane in host_attempts["lanes"] if lane["id"] == "h3.native-euler") host = host_attempts["host"] case = native["case"] environment = native["environment"] return { "baselines": { "native_euler": { "device": environment["device"], "dtype": environment["dtype"], "fingerprint": case["trace_fingerprint"], "max_abs_error": case["max_abs_error"], "mean_abs_error": case["mean_abs_error"], "status": case["status"], "tolerance": case["tolerance"], } }, "capability": {"level": "allow", "reasons": ["compatible"]}, "determinism": { "deterministic_rerun": case["deterministic_rerun"], "max_abs_error": case["max_abs_error"], "mean_abs_error": case["mean_abs_error"], "status": case["status"], "tolerance": case["tolerance"], "trace_fingerprint": case["trace_fingerprint"], }, "evidence": { "artifact": { "construction_fingerprint": metadata["artifact"]["construction_fingerprint"], "numerical_fingerprint": metadata["artifact"]["numerical_fingerprint"], }, "receipt_fingerprint": lane["receipt_fingerprint"], "source_ids": ["parity.native_euler", "host.known_good"], }, "execution": { "counts": case["counts"], "rng_ownership": { "model": "none", "sampler": "none", "schedule": "none", }, "status": "succeeded", }, "id": "host.h3.native-euler", "lane": "h3_native_euler", "model_weights_present": False, "profile": _profile( identifier=native["profile"]["id"], variant="Turbo", evidence="official", recipe="krea2.turbo.official-8", ), "repeat": _repeat(lane), "runtime": _runtime( device=environment["device"], dtype=environment["dtype"], python=environment["python"], torch=environment["torch"], numpy=environment["numpy"], comfyui=host["version"], comfyui_revision_chunks=host["revision_chunks"], sigmax_revision_chunks=host_attempts["sigmax_revision_chunks"], ), "schedule": {"image_seq_len": None, "mu": "1.15"}, "weight_variant": "none", "workload": _workload( requested_width=1024, requested_height=1024, effective_width=1024, effective_height=1024, transitions=8, ), } def build_matrix_envelope() -> dict[str, object]: sources = {path: _read_json(path) for path in SOURCE_PATHS} for path, source in sources.items(): if source.get("status") not in {None, "PASS"}: raise RuntimeError(f"{path} is not accepted evidence") results = _parity_rows( sources["tests/parity/fixtures/krea2_turbo_parity_v1.json"], sources["tests/parity/fixtures/krea2_raw_parity_v1.json"], ) results.extend(_anima_rows(sources["tests/golden/anima_v1.json"])) results.extend(_wan_rows(sources["tests/golden/wan_v1.json"])) results.extend(_ltx_rows(sources["tests/golden/ltx_v1.json"])) results.extend( _workflow_rows( sources["comfyui_sigmax/workflows/fixtures.json"], sources["tests/benchmarks/fixtures/known_good_host_attempts_v1.json"], sources["tests/parity/fixtures/krea2_raw_parity_v1.json"], ) ) results.append( _native_euler_row( sources["tests/parity/fixtures/krea2_native_euler_parity_v1.json"], sources["comfyui_sigmax/workflows/fixtures.json"], sources["tests/benchmarks/fixtures/known_good_host_attempts_v1.json"], ) ) results.sort(key=lambda row: cast(str, row["id"])) coverage = { lane: sum(row["lane"] == lane for row in results) for lane in ( "anima_schedule_parity", "h2_workflow", "h3_native_euler", "raw_schedule_parity", "turbo_schedule_parity", "wan_schedule_parity", "ltx_schedule_parity", ) } coverage["total_verified_results"] = len(results) matrix = { "coverage": coverage, "exclusions": [ "advanced_workflows", "gpu_model_weights", "partial_denoise_execution", "resume_execution", "stochastic_euler_execution", ], "results": results, "schema": "sigmax.numerical-benchmark-matrix/1", "sources": _source_records(sources), "weight_variants": [ { "kind": "bf16", "reason": "gpu_model_weights_not_approved", "result": None, "status": "not_evaluated", }, { "kind": "quantized", "reason": "gpu_model_weights_not_approved", "result": None, "status": "not_evaluated", }, ], } matrix_bytes = _canonical(matrix) return { "matrix": matrix, "matrix_fingerprint": _identity(matrix_bytes), "schema": "sigmax.numerical-benchmark-matrix-envelope/1", } def main() -> int: parser = argparse.ArgumentParser() mode = parser.add_mutually_exclusive_group(required=True) mode.add_argument("--check", action="store_true") mode.add_argument("--write", action="store_true") args = parser.parse_args() payload = _canonical(build_matrix_envelope()) + b"\n" if args.check: if not TARGET.is_file() or TARGET.read_bytes() != payload: print("BENCHMARK_MATRIX=FAIL") return 1 print("BENCHMARK_MATRIX=PASS") return 0 TARGET.parent.mkdir(parents=True, exist_ok=True) TARGET.write_bytes(payload) print(f"WROTE={TARGET.relative_to(ROOT).as_posix()}") return 0 if __name__ == "__main__": raise SystemExit(main())