Omniverse Usd Performance Tuning

references/optimization-report/scripts/validate_report.py

#!/usr/bin/env python3
# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0
"""Validate a USD Performance Tuning report against optimization-report.schema.json.

Deterministic local validation with no third-party runtime dependencies. The
agent (or CI) should run this before treating a report as final, so an
out-of-enum verdict, a missing required field, or an unexpected array-item key
is caught instead of shipping a schema-invalid report.

Implements the JSON Schema draft-07 subset this schema uses: type (including
type unions like ["string", "null"]), enum, required, properties,
additionalProperties=false, items, minimum, and maximum.

Phase-4 target coverage gate
----------------------------
Schema validation alone cannot catch a Phase-4 target that was never enumerated
in the report (the failure mode where an assembly_root remainder is silently
left un-optimized). A report's ``target_coverage.complete`` flag is self-attested
by the report author, so the gate reconciles ``target_coverage`` against the
upstream apply-restructure manifest(s): the report must cover the UNION of every
iteration's ``phase4_targets[]``, every disposition must be resolved, and
``skipped_zero_meshes`` is accepted only when the manifest's authoritative
``mesh_count`` for that target is 0.

Reconciliation is fail-closed, not opt-in: when any coverage entry has a
restructure role (assembly_root | prototype | shared_layer | loadable_subasset)
a manifest is REQUIRED. Manifests are taken from ``--manifest`` and/or the
report's own ``target_coverage.source_manifests[]`` (auto-loaded relative to the
report), so a restructure report cannot pass merely because the operator forgot
the flag. Monolith/diagnosis runs (no restructure roles) stay manifest-free.

Footprint gate (repack-normalization)
-------------------------------------
When the optional ``footprint`` block is present, the gate checks the
raw -> repack_normalized -> optimized arithmetic is consistent and fails closed
on a repack-as-optimization claim: a raw->optimized saving that is almost
entirely the free crate re-encode (structural shrink ~0 off the
repack-normalized baseline) cannot be presented as the optimization win.

Preservation gate (axis-A silent-loss)
--------------------------------------
When the optional ``preservation`` block is present, the gate checks its shape
matches what the run-scoring oracle consumes:
integer ``rendered_mesh_count`` / ``dangling`` and boolean
``distinct_geometry_bytes_preserved`` / ``bounds_preserved``. Whether those
values actually pass (count unchanged, bounds/bytes preserved, dangling 0) is
scored against the asset's oracle, not here.

Safety-gate gate (Phase-2c correctness precondition)
----------------------------------------------------
The ``safety_gate`` block records whether a correctness precondition — dangling
material bindings, a broken reference — was cleared before structuring. It was
optional and unchecked, so deleting it turned a run the scorer zeroes into one it
passed at 1.0: the gradient rewarded hiding the finding. This gate requires the
block whenever a ``validators[]`` row reports issues for a concept the registry
(``usd-validation-runner/references/validator-concepts.json``) tags
``role: safety_gate``, and requires a present block to state a non-empty status.
The status vocabulary is a schema enum, and the run-scoring oracle reads the same
``safety_gate_status`` helper, so an unstated gate is unresolved on both paths.

Score-arithmetic gate
---------------------
``optimization_score`` is defined as the weighted mean over ``metric_groups[]``
and ``score_label`` as a band derived from it. Both rules lived only in prose and
schema descriptions, so a report declaring 2.0 with the label ``strong`` — or a
weighted mean of 1.0 against a declared 8.0 — validated clean. Both are now
recomputed from data already in the report.

Descent-convergence gate (premature-merge / premature-decimate)
---------------------------------------------------------------
A run that did hierarchy dedup / instancing must not run a Phase-4 geometry op
(decimation, within-prototype merge, primitive-fit, quantization, topology edit)
before the bounded recursive de-duplication descent has CONVERGED. The skill
documents this descent and its convergence as prose plus the manifest fields
``frontier.descent_converged`` / ``frontier.final_rescan_new_groups_above_floor``,
but nothing enforced it. This gate fails closed, keyed on the output
report/manifest (so a hand-authored rewrite is gated the same as a native
``deduplicateHierarchies`` run), unless convergence is recorded in the manifest
frontier block or the report ``structural_summary`` mirror keys
(``frontier_descent_converged`` + ``frontier_final_rescan_new_groups_above_floor``
== 0), or an explicit ``frontier.convergence_waived_by_user`` is set. It also
rejects the ``descent_converged == true`` with residue-remaining contradiction.

Usage:
    python3 validate_report.py <report.json> [--schema <schema.json>] \\
        [--manifest <apply-restructure-manifest.json> ...] \\
        [--validator-concepts <validator-concepts.json>]
Exit code 0 when the report conforms and the coverage gate passes, 1 otherwise.
"""
from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path
from typing import Any

# The descent-convergence gate's pure decision functions + constants live in a
# sibling module (they ship together in this scripts dir). Adding our own
# directory to sys.path lets the plain ``import`` work when the validator is run
# as a standalone script rather than a package.
sys.path.insert(0, str(Path(__file__).resolve().parent))
import descent_convergence_common as dcc  # noqa: E402

sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
from component_scope import identity_change_authorized, retains_boundary, validate_scope

DEFAULT_SCHEMA = Path(__file__).resolve().parent / "optimization-report.schema.json"
#: The validator-concept registry is the single source of truth for which concepts
#: are safety gates (``role: safety_gate``). It ships in the sibling
#: usd-validation-runner reference; this validator reads it and never restates it.
DEFAULT_VALIDATOR_CONCEPTS = (
    Path(__file__).resolve().parents[2]
    / "usd-validation-runner"
    / "references"
    / "validator-concepts.json"
)

#: A Phase-4 target is "resolved" only with one of these dispositions. ``blocked``
#: (or a target with no entry at all) keeps ``target_coverage.complete`` false and
#: the report non-final — mirroring the validation report's RESOLVED_STATUSES.
PHASE4_RESOLVED_DISPOSITIONS = frozenset(
    {"optimized", "exact_cache_hit", "normalized_cache_hit", "product_identity_hit", "verified_unchanged", "skipped_zero_meshes", "skipped_user_declined"}
)
RESTRUCTURE_TARGET_CLASSES = frozenset(
    {"prototype", "shared_layer", "loadable_subasset", "assembly_root"}
)
#: Reuse/descent frontier dispositions that SHARE a unit (and therefore make a
#: scene-graph / consolidation claim that must be backed by MEASURED reuse).
SHARED_IDENTITY_DISPOSITIONS = frozenset({"externalize_shared", "internal_share"})
#: Reuse/descent frontier reduction routes that DESTROY identity — allowed only on weak-identity
#: (anonymous / structural-fallback) units, never on a strong-identity part.
IDENTITY_DESTROYING_ROUTES = frozenset({"point_instance", "merge"})
#: Identity signals that mark a unit as a real, addressable part (strong identity).
STRONG_IDENTITY_SIGNALS = frozenset({"kind", "naming", "semantic"})
#: The structural-fallback grain: identity is absent, so the coarsest repeating
#: subtree proposed the boundary. Weak identity, but a LEGAL shared-frontier basis —
#: this value records a declared structural grouping without product identity.
STRUCTURAL_FALLBACK_SIGNAL = "structural_fallback"
#: Every identity_signal the apply-restructure manifest schema admits. Anything
#: else (a typo, an invented spelling) is rejected rather than waved through: the
#: shared/destroying checks below key on exact values, so an unknown signal would
#: otherwise slip past both of them.
LEGAL_IDENTITY_SIGNALS = STRONG_IDENTITY_SIGNALS | {STRUCTURAL_FALLBACK_SIGNAL, "none"}
#: ``safety_gate.status`` values that mean the correctness precondition IS cleared.
SAFETY_GATE_CLEARED_STATUSES = frozenset(
    {"resolved", "clear", "none", "passed", "not_required"}
)
#: ...and the values that say it is not. Together these are the schema enum, so a
#: typo cannot land in the gap between "cleared" and "uncleared" and read as clean.
SAFETY_GATE_UNCLEARED_STATUSES = frozenset({"unresolved", "blocked", "failed"})
SAFETY_GATE_STATUSES = SAFETY_GATE_CLEARED_STATUSES | SAFETY_GATE_UNCLEARED_STATUSES
#: Score bands, high to low, as ``(label, inclusive_lower_bound)``. Mirrors the
#: schema's score_label description and the README's derivation rule.
SCORE_BANDS: tuple[tuple[str, float], ...] = (
    ("excellent", 9.0),
    ("strong", 7.5),
    ("moderate", 5.5),
    ("neutral", 4.5),
    ("mixed", 2.5),
    ("regressed", float("-inf")),
)
#: How far the declared ``optimization_score`` may sit from the recomputed weighted
#: mean. The contract rounds the mean to one decimal, so 0.05 admits every correct
#: rounding and nothing else.
_SCORE_TOLERANCE = 0.05
#: Coverage-entry roles that mean "a restructure happened", so a manifest is
#: mandatory and reconciliation is not optional. The ``monolith`` role (an
#: optimize-as-is N=1 target) and an empty ledger stay manifest-free. Shared with
#: the descent-convergence gate (dcc.RESTRUCTURE_ROLES) so the two stay in lockstep.
RESTRUCTURE_ROLES = dcc.RESTRUCTURE_ROLES


def _is_number(value: Any) -> bool:
    """True for a real JSON number. Python counts a bool as an int; JSON does not."""
    return isinstance(value, (int, float)) and not isinstance(value, bool)


def _type_ok(instance: Any, type_name: str) -> bool:
    if type_name == "object":
        return isinstance(instance, dict)
    if type_name == "array":
        return isinstance(instance, list)
    if type_name == "string":
        return isinstance(instance, str)
    if type_name == "number":
        return isinstance(instance, (int, float)) and not isinstance(instance, bool)
    if type_name == "integer":
        return isinstance(instance, int) and not isinstance(instance, bool)
    if type_name == "boolean":
        return isinstance(instance, bool)
    if type_name == "null":
        return instance is None
    return True


def _validate(instance: Any, schema: dict, path: str, errors: list[str]) -> None:
    declared_type = schema.get("type")
    if declared_type is not None:
        candidates = declared_type if isinstance(declared_type, list) else [declared_type]
        if not any(_type_ok(instance, name) for name in candidates):
            got = "null" if instance is None else type(instance).__name__
            errors.append(f"{path}: expected type {candidates}, got {got}")
            return

    if "enum" in schema and instance not in schema["enum"]:
        errors.append(f"{path}: {instance!r} is not one of {schema['enum']}")

    if isinstance(instance, (int, float)) and not isinstance(instance, bool):
        if "minimum" in schema and instance < schema["minimum"]:
            errors.append(f"{path}: {instance} is below minimum {schema['minimum']}")
        if "maximum" in schema and instance > schema["maximum"]:
            errors.append(f"{path}: {instance} is above maximum {schema['maximum']}")

    if isinstance(instance, dict):
        properties = schema.get("properties", {})
        for required_key in schema.get("required", []):
            if required_key not in instance:
                errors.append(f"{path}: missing required property '{required_key}'")
        allow_additional = schema.get("additionalProperties", True)
        for key, value in instance.items():
            if key in properties:
                _validate(value, properties[key], f"{path}.{key}", errors)
            elif allow_additional is False:
                errors.append(f"{path}: unexpected property '{key}'")

    if isinstance(instance, list) and "items" in schema:
        for index, item in enumerate(instance):
            _validate(item, schema["items"], f"{path}[{index}]", errors)


def validate_report(report: Any, schema: dict | None = None) -> list[str]:
    """Return a list of schema-violation messages; empty list means the report conforms."""
    if schema is None:
        schema = json.loads(DEFAULT_SCHEMA.read_text(encoding="utf-8"))
    errors: list[str] = []
    _validate(report, schema, "$", errors)
    return errors


def _validate_frontier_entry(target: dict, label: str) -> list[str]:
    """Check boundary evidence and scope without requiring the USD runtime.

    A retained interior merge preserves the named parent. Unscoped merges and
    point-instancer conversion retain the historical identity-change guard.
    Legacy manifests without additive scope fields remain readable.
    """
    errors: list[str] = []
    disposition = target.get("identity_disposition")
    route = target.get("reduction_route")
    signal = target.get("identity_signal")
    intent_confirmed = target.get("intent_confirmed") is True or identity_change_authorized(target)

    is_shared = disposition in SHARED_IDENTITY_DISPOSITIONS
    errors.extend(f'{label}: {error}' for error in validate_scope(target))
    is_destroying = route in IDENTITY_DESTROYING_ROUTES and not retains_boundary(target)

    if (is_shared or is_destroying) and not signal:
        errors.append(
            f"{label}: identity_signal is required on a shared "
            f"({disposition}) or identity-destroying ({route}) entry "
            "(the boundary basis must be queryable)"
        )
    elif signal is not None and signal not in LEGAL_IDENTITY_SIGNALS:
        errors.append(
            f"{label}: identity_signal {signal!r} is not one of "
            f"{sorted(LEGAL_IDENTITY_SIGNALS)} (the apply-restructure manifest enum) "
            "— an unrecognised signal would slip past both identity guards"
        )

    if is_shared and signal == "none":
        errors.append(
            f"{label}: identity_disposition {disposition!r} landed on anonymous "
            "meshes (identity_signal 'none') — a shared frontier must land on a "
            "named/kind/semantic unit or, at worst, the coarsest repeating subtree "
            "(identity_signal 'structural_fallback'), never anonymous geometry"
        )

    if is_destroying and signal in STRONG_IDENTITY_SIGNALS and not intent_confirmed:
        errors.append(
            f"{label}: identity-destroying reduction_route {route!r} on a "
            f"strong-identity unit (identity_signal {signal!r}) — point_instance / "
            "merge are reserved for weak-identity (anonymous / structural_fallback) "
            "units; a named/kind/semantic part must stay addressable unless the "
            "entry records an explicit intent_confirmed: true"
        )

    return errors


def validate_manifest_structure(manifest: Any) -> list[str]:
    """Enforce the load-bearing apply-restructure manifest invariants.

    Independent of the JSON-Schema walker so the rules hold without ``jsonschema``:
    a ``mode=restructure`` manifest must carry a non-empty ``phase4_targets[]``,
    and every target must declare an integer ``mesh_count >= 0`` (the authoritative
    default-predicate count the coverage gate keys on).
    """
    errors: list[str] = []
    if not isinstance(manifest, dict):
        return [f"manifest must be an object, got {type(manifest).__name__}"]
    mode = manifest.get("mode")
    targets = manifest.get("phase4_targets")
    if mode == "restructure" and not targets:
        errors.append(
            "mode=restructure manifest must list a non-empty phase4_targets[] "
            "(do not drop the key; an assembly_root with retained meshes must appear)"
        )
    for index, target in enumerate(targets or []):
        where = f"phase4_targets[{index}]"
        path = target.get("path") if isinstance(target, dict) else None
        label = f"{where} ({path})" if path else where
        if not isinstance(target, dict):
            errors.append(f"{where}: must be an object")
            continue
        if not isinstance(path, str) or not path:
            errors.append(f"{where}: missing required 'path'")
        target_class = target.get("target_class")
        if target_class not in RESTRUCTURE_TARGET_CLASSES:
            errors.append(
                f"{label}: target_class {target_class!r} not in {sorted(RESTRUCTURE_TARGET_CLASSES)}"
            )
        mesh_count = target.get("mesh_count")
        if isinstance(mesh_count, bool) or not isinstance(mesh_count, int) or mesh_count < 0:
            errors.append(
                f"{label}: mesh_count must be an integer >= 0 (authoritative "
                f"default-predicate count), got {mesh_count!r}"
            )
        errors.extend(_validate_frontier_entry(target, label))

    # Measured-reuse-before-consolidation (C3): a scene-graph / consolidation win
    # may only be reported from MEASURED reuse, never estimated. When any target
    # claims a shared disposition (externalize_shared / internal_share), the
    # manifest's top-level ``frontier`` block must record reuse_measured: true.
    shared_entries = [
        t for t in (targets or [])
        if isinstance(t, dict) and t.get("identity_disposition") in SHARED_IDENTITY_DISPOSITIONS
    ]
    if shared_entries:
        frontier = manifest.get("frontier")
        if not isinstance(frontier, dict) or frontier.get("reuse_measured") is not True:
            errors.append(
                "frontier: a shared disposition (externalize_shared/internal_share) is claimed "
                "but frontier.reuse_measured is not true — a scene-graph / consolidation win "
                "must be backed by MEASURED reuse (distinct vs total units), never estimated. "
                "When measured reuse is low, pivot to the disk tier instead of forcing sharing."
            )

    # apply-restructure residual-mesh postcondition (mirrored here for the
    # manifest, enforced live in apply-restructure where the USD stage is open).
    # When extraction leaves > 0 mesh prims on the assembly root, apply-restructure
    # records the authoritative residual count under ``assembly_root`` AND lists the
    # same root in ``phase4_targets[]`` as ``target_class: assembly_root``. Fail loud
    # if a restructure manifest is written with residual meshes but no such target
    # entry: a retained-mesh assembly root must never be silently dropped from Phase 4.
    residual = manifest.get("assembly_root")
    if isinstance(residual, dict):
        residual_count = residual.get("mesh_count")
        residual_path = residual.get("path")
        if (
            isinstance(residual_count, int)
            and not isinstance(residual_count, bool)
            and residual_count > 0
        ):
            root_entry = next(
                (
                    t
                    for t in (targets or [])
                    if isinstance(t, dict)
                    and t.get("target_class") == "assembly_root"
                    and (residual_path is None or t.get("path") == residual_path)
                ),
                None,
            )
            if root_entry is None:
                errors.append(
                    f"assembly_root records residual mesh_count {residual_count} > 0 but no "
                    "phase4_targets[] entry with target_class 'assembly_root'"
                    + (f" and path {residual_path}" if residual_path else "")
                    + " is present (apply-restructure postcondition: a retained-mesh assembly "
                    "root must be a Phase-4 target, never silently dropped)"
                )
            elif root_entry.get("mesh_count") != residual_count:
                errors.append(
                    f"assembly_root residual mesh_count {residual_count} does not match its "
                    f"phase4_targets[] entry mesh_count {root_entry.get('mesh_count')!r} "
                    "(both must echo the same authoritative default-predicate count)"
                )
    return errors


def load_recorded_manifests(
    report: Any, base_dir: Path
) -> tuple[list[tuple[str, Any]], list[str]]:
    """Load the manifests recorded in ``target_coverage.source_manifests[]``.

    Relative paths resolve against ``base_dir`` (the report's directory) so a
    report can carry its own provenance and the gate fails closed without the
    operator having to remember ``--manifest``. Returns ``(labeled_manifests,
    errors)`` where each labeled manifest is ``(source_path, manifest_dict)``.
    """
    labeled: list[tuple[str, Any]] = []
    errors: list[str] = []
    coverage = report.get("target_coverage") if isinstance(report, dict) else None
    if not isinstance(coverage, dict):
        return labeled, errors
    for rel in coverage.get("source_manifests", []) or []:
        path = Path(rel)
        if not path.is_absolute():
            path = base_dir / path
        try:
            labeled.append((rel, json.loads(path.read_text(encoding="utf-8"))))
        except (OSError, json.JSONDecodeError) as exc:
            errors.append(
                f"target_coverage.source_manifests entry {rel!r} could not be loaded: {exc}"
            )
    return labeled, errors


def _coverage_key(record):
    return (record["path"], record.get("definition_id", ""))


def _manifest_targets(manifests: list[Any]) -> dict[tuple[str, str], int | None]:
    """Union of each layer/definition identity and its authoritative mesh count.

    Multi-iteration runs must reconcile against the UNION: the exact regression
    that prompted this gate was iteration 1 listing an assembly_root that
    iteration 2's manifest dropped, leaving it uncovered by the final report.
    """
    planned: dict[tuple[str, str], int | None] = {}
    for manifest in manifests:
        for target in manifest.get("phase4_targets", []) or []:
            if isinstance(target, dict) and isinstance(target.get("path"), str):
                planned[_coverage_key(target)] = target.get("mesh_count")
    return planned


#: Arithmetic-consistency tolerance (percentage points) for the footprint split.
_FOOTPRINT_TOLERANCE_PP = 0.6
#: A structural change smaller than this magnitude (percent) is treated as "no
#: real structural shrink off the repack-normalized baseline" for the
#: repack-as-optimization fail-closed check.
_FOOTPRINT_STRUCTURAL_EPSILON_PCT = 1.0


def _pct_change(before: float, after: float) -> float | None:
    if before == 0:
        return None
    return (after - before) / before * 100.0


def validate_footprint(report: Any) -> list[str]:
    """Gate the optional repack-normalized footprint block.

    Returns violation messages (empty == passes or no footprint block). When a
    ``footprint`` block is present it must (1) be arithmetically consistent
    across raw -> repack_normalized -> optimized, and (2) fail closed if the
    only reduction is the free crate re-encode while the structural win off the
    repack-normalized baseline is ~0 (repack-as-optimization). A repack — or an
    unshared disaggregation — presented as the optimization win is a fail-closed
    reporting error per the plan.
    """
    errors: list[str] = []
    footprint = report.get("footprint") if isinstance(report, dict) else None
    if footprint is None:
        return errors
    if not isinstance(footprint, dict):
        return ["footprint must be an object"]

    raw = footprint.get("raw_input_bytes")
    normalized = footprint.get("repack_normalized_baseline_bytes")
    optimized = footprint.get("optimized_bytes")
    nums = {
        "raw_input_bytes": raw,
        "repack_normalized_baseline_bytes": normalized,
        "optimized_bytes": optimized,
    }
    for name, value in nums.items():
        if not isinstance(value, (int, float)) or isinstance(value, bool) or value < 0:
            errors.append(f"footprint.{name} must be a number >= 0, got {value!r}")
    if errors:
        return errors

    if footprint.get("scored_against") != "repack_normalized":
        errors.append(
            "footprint.scored_against must be 'repack_normalized' — the storage "
            "dimension and any footprint claim score against the repack-normalized "
            "baseline, not the raw input"
        )

    structural_delta = footprint.get("structural_delta_pct")
    expected_structural = _pct_change(normalized, optimized)
    if isinstance(structural_delta, (int, float)) and not isinstance(structural_delta, bool):
        if expected_structural is not None and abs(structural_delta - expected_structural) > _FOOTPRINT_TOLERANCE_PP:
            errors.append(
                f"footprint.structural_delta_pct ({structural_delta}) does not match "
                f"repack_normalized->optimized ({expected_structural:.2f}%) within "
                f"{_FOOTPRINT_TOLERANCE_PP}pp"
            )
    else:
        errors.append("footprint.structural_delta_pct must be a number")

    repack_delta = footprint.get("repack_delta_pct")
    if isinstance(repack_delta, (int, float)) and not isinstance(repack_delta, bool):
        expected_repack = _pct_change(raw, normalized)
        if expected_repack is not None and abs(repack_delta - expected_repack) > _FOOTPRINT_TOLERANCE_PP:
            errors.append(
                f"footprint.repack_delta_pct ({repack_delta}) does not match "
                f"raw->repack_normalized ({expected_repack:.2f}%) within "
                f"{_FOOTPRINT_TOLERANCE_PP}pp"
            )

    raw_delta = footprint.get("raw_delta_pct")
    if isinstance(raw_delta, (int, float)) and not isinstance(raw_delta, bool):
        expected_raw = _pct_change(raw, optimized)
        if expected_raw is not None and abs(raw_delta - expected_raw) > _FOOTPRINT_TOLERANCE_PP:
            errors.append(
                f"footprint.raw_delta_pct ({raw_delta}) does not match "
                f"raw->optimized ({expected_raw:.2f}%) within {_FOOTPRINT_TOLERANCE_PP}pp"
            )

    # Repack-as-optimization fail-closed: a raw->optimized saving that is almost
    # entirely the free crate re-encode (structural shrink ~0 off the normalized
    # baseline) must not be presented as the optimization win.
    if (
        isinstance(structural_delta, (int, float))
        and not isinstance(structural_delta, bool)
        and structural_delta > -_FOOTPRINT_STRUCTURAL_EPSILON_PCT
    ):
        observed_raw = raw_delta if isinstance(raw_delta, (int, float)) and not isinstance(raw_delta, bool) else _pct_change(raw, optimized)
        if observed_raw is not None and observed_raw < -_FOOTPRINT_STRUCTURAL_EPSILON_PCT:
            storage_metrics=[row for row in report.get('metrics',[]) if isinstance(row,dict)
                             and row.get('category')=='storage_proxy' and row.get('name') not in ('raw_referenced_storage_bytes', 'scene_storage_bytes', 'reuse_overhead_bytes')]
            storage_groups=[row for row in report.get('metric_groups',[]) if isinstance(row,dict) and row.get('id')=='storage_proxy']
            honest=(any(row.get('before')==normalized and row.get('after')==optimized and row.get('verdict') in ('neutral','regressed')
                        for row in storage_metrics)
                    and isinstance(report.get('optimization_score'),(int,float)) and report['optimization_score']<=5
                    and storage_groups and all(isinstance(row.get('score'),(int,float)) and row['score']<=5 for row in storage_groups))
            if honest:
                return errors
            errors.append(
                f"footprint: raw->optimized shows {observed_raw:.2f}% but the structural "
                f"reduction off the repack-normalized baseline is only {structural_delta:.2f}% "
                "— the saving is the free crate re-encode, not the optimization. Presenting a "
                "repack (or an unshared disaggregation) as the optimization win is a "
                "fail-closed reporting error."
            )
    return errors


def validate_preservation(report: Any) -> list[str]:
    """Gate the optional axis-A preservation (silent-loss) block.

    Returns violation messages (empty == passes or no preservation block). The
    block is optional at the report level, but when present its shape must match
    exactly what the run-scoring oracle consumes so the
    schema, scorer, and validator agree: an integer ``rendered_mesh_count`` and
    ``dangling`` (both >= 0), and boolean ``distinct_geometry_bytes_preserved``
    and ``bounds_preserved``. This is shape-only — whether the values pass the
    gate (count unchanged, both bools true, dangling 0) is the scorer's job
    against the asset's oracle, not a static check here.
    """
    errors: list[str] = []
    pres = report.get("preservation") if isinstance(report, dict) else None
    if pres is None:
        return errors
    if not isinstance(pres, dict):
        return ["preservation must be an object"]
    for name in ("rendered_mesh_count", "dangling"):
        value = pres.get(name)
        if isinstance(value, bool) or not isinstance(value, int) or value < 0:
            errors.append(
                f"preservation.{name} must be an integer >= 0, got {value!r}"
            )
    for name in ("distinct_geometry_bytes_preserved", "bounds_preserved"):
        value = pres.get(name)
        if not isinstance(value, bool):
            errors.append(f"preservation.{name} must be a boolean, got {value!r}")
    return errors


class ValidatorConceptsUnavailable(RuntimeError):
    """The validator-concept registry could not be read.

    Raised rather than swallowed: the rule the safety-gate check exists to enforce
    is that silence must not read as a clean gate, so a check that cannot run has
    to say so instead of passing.
    """


_SAFETY_GATE_CONCEPTS_CACHE: dict[str, frozenset[str]] = {}


def load_safety_gate_concepts(concepts_path: Path | None = None) -> frozenset[str]:
    """Canonical names of every registry concept tagged ``role: safety_gate``.

    The registry (``usd-validation-runner/references/validator-concepts.json``) is
    the single source of truth for that tag; this validator reads it rather than
    keeping a second copy of the list that could drift.
    """
    path = Path(concepts_path) if concepts_path is not None else DEFAULT_VALIDATOR_CONCEPTS
    key = str(path)
    cached = _SAFETY_GATE_CONCEPTS_CACHE.get(key)
    if cached is not None:
        return cached
    try:
        registry = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        raise ValidatorConceptsUnavailable(
            f"validator-concept registry {path} could not be read ({exc}), so the "
            "safety-gate check cannot tell which validators are safety gates"
        ) from exc
    concepts = registry.get("concepts") if isinstance(registry, dict) else None
    if not isinstance(concepts, list):
        raise ValidatorConceptsUnavailable(
            f"validator-concept registry {path} has no concepts[] array"
        )
    names = frozenset(
        concept["canonical_name"]
        for concept in concepts
        if isinstance(concept, dict)
        and concept.get("role") == "safety_gate"
        and isinstance(concept.get("canonical_name"), str)
    )
    if not names:
        raise ValidatorConceptsUnavailable(
            f"validator-concept registry {path} declares no role 'safety_gate' concepts"
        )
    _SAFETY_GATE_CONCEPTS_CACHE[key] = names
    return names


def fired_safety_gate_validators(
    report: Any, concepts_path: Path | None = None
) -> list[str]:
    """``validators[]`` row names that report issues for a registry safety gate.

    The registry is read only when at least one row reports issues, so a report
    with no validator findings never depends on it.
    """
    rows = report.get("validators") if isinstance(report, dict) else None
    if not isinstance(rows, list):
        return []
    reporting: list[str] = []
    for row in rows:
        if not isinstance(row, dict):
            continue
        name = row.get("name")
        issues = row.get("issues")
        if isinstance(name, str) and _is_number(issues) and issues > 0:
            reporting.append(name)
    if not reporting:
        return []
    gates = load_safety_gate_concepts(concepts_path)
    return sorted({name for name in reporting if name in gates})


def safety_gate_status(report: Any) -> str | None:
    """Normalized ``safety_gate.status``.

    ``None`` when the block is absent; ``""`` when the block is there but records
    no usable status (key missing, empty, or not a string). Both callers — this
    validator and the run-scoring oracle — read the state through this one
    function so they cannot disagree about what "unstated" means.
    """
    block = report.get("safety_gate") if isinstance(report, dict) else None
    if not isinstance(block, dict):
        return None
    raw = block.get("status")
    return raw.strip().lower() if isinstance(raw, str) else ""


def validate_safety_gate(report: Any, concepts_path: Path | None = None) -> list[str]:
    """Gate the Phase-2c ``safety_gate`` block: presence and a stated status.

    Two rules, both fail-closed:

    * the block is REQUIRED whenever a ``validators[]`` row reports issues for a
      concept the registry tags ``role: safety_gate``. Without this, deleting the
      block turned a run the scorer zeroes into a run it passes at 1.0 — the
      scoring gradient pointed at omission, which on an eval loop trains an agent
      to hide correctness findings;
    * a block that is present must record a non-empty ``status``. The set of legal
      status strings is the schema enum (see ``SAFETY_GATE_STATUSES``); this
      function checks only that something was stated.
    """
    errors: list[str] = []
    if not isinstance(report, dict):
        return errors
    try:
        fired = fired_safety_gate_validators(report, concepts_path)
    except ValidatorConceptsUnavailable as exc:
        return [f"safety_gate: {exc}"]

    status = safety_gate_status(report)
    if status is None:
        if fired:
            errors.append(
                "safety_gate block is missing but safety-gate validator(s) "
                f"{fired} report issues — a fired correctness gate must be recorded "
                "with its status (resolved/clear/none/passed/not_required when it was "
                "cleared, unresolved/blocked/failed when it was not). Omitting the "
                "block hides the finding from the scorer instead of clearing the gate."
            )
    elif not status:
        errors.append(
            "safety_gate is present but records no status — an unstated gate is "
            "treated as unresolved, never as cleared. Set safety_gate.status to one "
            f"of {sorted(SAFETY_GATE_STATUSES)}."
        )
    return errors


SCORE_INTERPOLATION = (
    "Piecewise linear storage score: reductions of 10%, 25%, 50%, 75% map to "
    "6, 7, 8.5, 10; changes within ±5% map to 5. Regression convention: "
    "increases of 10%, 20%, 100% map to 3, 1, 0. This proxy does not measure usability."
)


def score_from_change(change_pct: float) -> float:
    """Score a lower-is-better measurement using the documented report anchors."""
    import math
    if not math.isfinite(change_pct):
        raise ValueError('A score requires a finite measured percentage change')
    anchors = [(-75., 10.), (-50., 8.5), (-25., 7.), (-10., 6.),
               (-5., 5.), (5., 5.), (10., 3.), (20., 1.), (100., 0.)]
    if change_pct <= anchors[0][0]:
        return anchors[0][1]
    for (left, first), (right, last) in zip(anchors, anchors[1:]):
        if change_pct <= right:
            return round(first + (last-first)*(change_pct-left)/(right-left), 1)
    return anchors[-1][1]


def score_band(score: float) -> str:
    """The documented ``score_label`` band for a numeric ``optimization_score``."""
    for label, lower_bound in SCORE_BANDS:
        if score >= lower_bound:
            return label
    return SCORE_BANDS[-1][0]


def validate_score_arithmetic(report: Any) -> list[str]:
    """Gate the headline ``optimization_score`` and its ``score_label``.

    Both rules are stated in the schema descriptions and the report README, and
    until now neither was checked: a report could declare 2.0 with the label
    ``strong``, or a weighted mean of 1.0 against a declared 8.0, and validate
    clean. The score is the number every reader anchors on, and both inputs are
    already in the report, so recompute it:

    * the weighted mean over ``metric_groups[]`` with a non-null score and a
      non-zero weight must be within 0.05 of ``optimization_score`` (the contract
      rounds to one decimal, so that tolerance admits every correct rounding);
    * ``score_label`` must be the band the declared score falls in.
    """
    errors: list[str] = []
    if not isinstance(report, dict):
        return errors
    declared = report.get("optimization_score")
    groups = report.get("metric_groups")

    if _is_number(declared) and isinstance(groups, list):
        weighted_total = 0.0
        weight_total = 0.0
        negative_weights: list[Any] = []
        for group in groups:
            if not isinstance(group, dict):
                continue
            group_score = group.get("score")
            weight = group.get("weight")
            if not _is_number(weight):
                continue
            if weight < 0:
                negative_weights.append(group.get("id", "<unknown>"))
                continue
            if group_score is None or not _is_number(group_score) or weight == 0:
                continue
            weighted_total += group_score * weight
            weight_total += weight
        if negative_weights:
            errors.append(
                f"metric_groups {negative_weights} declare a negative weight; weights "
                "are a non-negative relative scale, and a negative one corrupts the "
                "weighted mean the top-level score is computed from"
            )
        if weight_total:
            computed = weighted_total / weight_total
            if abs(declared - computed) > _SCORE_TOLERANCE:
                errors.append(
                    f"optimization_score {declared} does not match the weighted mean of "
                    f"metric_groups[] ({computed:.4f}, rounds to {round(computed, 1)}) "
                    f"within {_SCORE_TOLERANCE}. The score is computed as "
                    "round(sum(score * weight) / sum(weight), 1) over groups with a "
                    "non-null score and a non-zero weight; do not hand-edit it afterwards."
                )
        elif not negative_weights:
            errors.append(
                f"optimization_score {declared} is declared but metric_groups[] has no "
                "scored group (non-null score with a non-zero weight) to compute it "
                "from; the headline score is derived from the groups, so at least one "
                "must carry a score"
            )

    label = report.get("score_label")
    if _is_number(declared) and isinstance(label, str):
        expected = score_band(declared)
        if label != expected:
            errors.append(
                f"score_label {label!r} contradicts optimization_score {declared}, "
                f"which falls in the {expected!r} band (excellent >= 9.0; strong >= 7.5; "
                "moderate >= 5.5; neutral >= 4.5; mixed >= 2.5; regressed < 2.5)"
            )
    return errors


def reconcile_target_coverage(report: Any, manifests: list[Any] | None = None) -> list[str]:
    """Gate the report's Phase-4 target_coverage; reconcile against manifest(s).

    Returns violation messages (empty == the gate passes). Always checks the
    report's internal consistency (resolved dispositions, the
    ``skipped_zero_meshes => mesh_count == 0`` rule, and the ``complete`` flag).
    When ``manifests`` are supplied it also asserts the covered set equals the
    union of every manifest's ``phase4_targets[]`` and cross-checks each
    disposition against the manifest's authoritative ``mesh_count``.
    """
    errors: list[str] = []
    coverage = report.get("target_coverage") if isinstance(report, dict) else None
    if not isinstance(coverage, dict):
        return ["target_coverage missing or not an object"]
    entries = coverage.get("entries", [])
    by_path: dict[tuple[str, str], dict[str, Any]] = {}
    for entry in entries:
        if isinstance(entry, dict) and isinstance(entry.get("path"), str):
            key = _coverage_key(entry)
            if key in by_path:
                errors.append(f"duplicate target_coverage definition entry: {key}")
            by_path[key] = entry

    checker = None
    native_contract = any(m.get('schema_version') == 2 for m in (manifests or []))
    for entry in entries:
        path = entry.get("path", "<unknown>")
        disposition = entry.get("disposition")
        mesh_count = entry.get("mesh_count")
        if disposition == "skipped_zero_meshes" and mesh_count != 0:
            errors.append(
                f"target_coverage entry {path}: skipped_zero_meshes requires "
                f"mesh_count == 0, got {mesh_count!r} (a non-zero target cannot be skipped)"
            )
        if disposition == "skipped_zero_meshes" and entry.get("geometry_count", mesh_count) != 0:
            errors.append(f"target_coverage entry {path}: zero meshes does not establish empty geometry")
        if disposition == "verified_unchanged" or (native_contract and disposition == "optimized"):
            if checker is None:
                import importlib.util
                checker_spec = importlib.util.spec_from_file_location("report_conformance", Path(__file__).with_name("structure_conformance.py"))
                checker = importlib.util.module_from_spec(checker_spec)
                checker_spec.loader.exec_module(checker)
            for error in checker.check_definition_result(entry, require_categories=native_contract):
                errors.append(f"target_coverage entry {path}: {error}")
        # No-op-masquerading-as-optimized: an entry that claims the mesh op chain
        # ran must have touched at least one mesh when the target has meshes.
        optimized_mesh_count = entry.get("optimized_mesh_count")
        if (
            disposition == "optimized"
            and isinstance(optimized_mesh_count, int)
            and not isinstance(optimized_mesh_count, bool)
            and optimized_mesh_count == 0
            and isinstance(mesh_count, int)
            and not isinstance(mesh_count, bool)
            and mesh_count > 0
        ):
            errors.append(
                f"target_coverage entry {path}: disposition 'optimized' but "
                f"optimized_mesh_count is 0 while mesh_count is {mesh_count} > 0 "
                "(no-op masquerading as optimized — record the meshes actually touched, "
                "or use skipped_user_declined / skipped_zero_meshes)"
            )

    present_restructure_roles = sorted(
        {e.get("role") for e in entries} & RESTRUCTURE_ROLES
    )

    # Once a restructure exists (a manifest is supplied/recorded OR a restructure
    # role appears), 'monolith' is illegal: it is by definition the N=1, no-manifest,
    # non-restructured optimize-as-is path. A monolith entry alongside manifest
    # provenance or restructure roles is a contradiction, not a valid ledger row.
    recorded_manifests = coverage.get("source_manifests") or []
    manifest_context = bool(manifests) or bool(recorded_manifests)
    if manifest_context or present_restructure_roles:
        for entry in entries:
            if entry.get("role") == "monolith":
                errors.append(
                    f"target_coverage entry {entry.get('path', '<unknown>')}: role 'monolith' "
                    "is illegal once a restructure exists (a source manifest is present or a "
                    "restructure role appears in target_coverage). 'monolith' is only the N=1, "
                    "no-manifest, non-restructured optimize-as-is target."
                )

    if present_restructure_roles and not manifests:
        errors.append(
            "target_coverage has restructure role(s) "
            f"{present_restructure_roles} but no source manifest was supplied or recorded; "
            "reconciliation is mandatory once a restructure happened. Record "
            "target_coverage.source_manifests[] (or pass --manifest) so the covered set is "
            "reconciled against the planned phase4_targets[] instead of self-attested."
        )

    all_resolved = all(
        e.get("disposition") in PHASE4_RESOLVED_DISPOSITIONS for e in entries
    )
    if coverage.get("complete") is not True:
        errors.append(
            "target_coverage.complete must be true for a final report "
            "(false => a Phase-4 target is unresolved/blocked)"
        )
    elif not all_resolved:
        errors.append(
            "target_coverage.complete is true but some entries are unresolved "
            "(only optimized | exact_cache_hit | normalized_cache_hit | product_identity_hit | verified_unchanged | skipped_zero_meshes | skipped_user_declined count as resolved)"
        )

    if manifests:
        planned = _manifest_targets(manifests)
        planned_paths = set(planned)
        covered_paths = set(by_path)
        for path in sorted(planned_paths - covered_paths):
            errors.append(
                f"target_coverage is missing an entry for manifest phase4_target: {path} "
                "(every planned Phase-4 target, across all iterations, must be covered)"
            )
        for path in sorted(covered_paths - planned_paths):
            errors.append(
                f"target_coverage entry {path} is not present in any supplied manifest "
                "phase4_targets[] (unexpected target or a missing manifest)"
            )
        for path in sorted(planned_paths & covered_paths):
            authoritative = planned[path]
            disposition = by_path[path].get("disposition")
            if (
                disposition == "skipped_zero_meshes"
                and isinstance(authoritative, int)
                and authoritative > 0
            ):
                errors.append(
                    f"target_coverage entry {path}: skipped_zero_meshes but the manifest's "
                    f"authoritative mesh_count is {authoritative} > 0 (lying skip)"
                )
    return errors


def _collect_frontier_states(report: Any, manifests: list[Any]) -> list[tuple[str, dict]]:
    """Gather ``(source_label, frontier_dict)`` from every manifest ``frontier`` block
    and from the report's ``structural_summary`` mirror keys, so convergence can be
    read from the manifest OR the report (a manifest-less run carries it in the report)."""
    states: list[tuple[str, dict]] = []
    for i, manifest in enumerate(manifests or []):
        if isinstance(manifest, dict) and isinstance(manifest.get("frontier"), dict):
            states.append((f"manifest[{i}].frontier", manifest["frontier"]))
    if isinstance(report, dict):
        ss = report.get("structural_summary")
        if isinstance(ss, dict):
            mirror: dict[str, Any] = {}
            if "frontier_descent_converged" in ss:
                mirror["descent_converged"] = ss.get("frontier_descent_converged")
            if "frontier_final_rescan_new_groups_above_floor" in ss:
                mirror["final_rescan_new_groups_above_floor"] = ss.get(
                    "frontier_final_rescan_new_groups_above_floor"
                )
            if ss.get("frontier_convergence_waived"):
                mirror["convergence_waived_by_user"] = ss.get("frontier_convergence_waived")
                mirror["convergence_waiver_reason"] = ss.get(
                    "frontier_convergence_waiver_reason"
                )
            if mirror:
                states.append(("structural_summary", mirror))
    return states


def validate_descent_convergence(report: Any, manifests: list[Any] | None = None) -> list[str]:
    """Gate: a Phase-4 geometry op must not run on an UNCONVERGED dedup frontier.

    FF1 — if the run did hierarchy dedup / instancing AND ran a Phase-4 geometry op
    (decimation, within-prototype merge, primitive-fit, quantization, topology edit),
    the manifest ``frontier`` block OR the report ``structural_summary`` mirror must
    record ``descent_converged == true`` with a dry terminal re-scan (or an explicit
    recorded waiver with a reason). Otherwise the run may have stopped at a coarse
    frontier and reduced geometry that further sharing would have collapsed
    (premature-merge / premature-decimate inflation).

    The gate keys on THIS-RUN dedup signals: an unambiguous dedup/instancing op
    marker, an apply-restructure manifest, an ``instance_count`` increase, a
    restructure-role coverage entry, or a measured-reuse flag (see
    ``dcc.did_dedup``). It is therefore only as good as those recorded signals: a
    run that dedup'd but recorded NONE of them is not detectable here (there is no
    absolute post-state that proves dedup happened this run without one of these
    signals). It is mechanism-agnostic across the signals it CAN see — a
    hand-authored rewrite that records a dedup op-marker / manifest / instance
    increase is gated the same as a native ``deduplicateHierarchies`` run.

    FF2 — ``descent_converged == true`` together with
    ``final_rescan_new_groups_above_floor > 0`` is the self-contradiction the
    apply-restructure manifest schema names but never enforces; fail it.
    """
    errors: list[str] = []
    manifests = manifests or []
    states = _collect_frontier_states(report, manifests)

    # FF2 — converged-but-residue contradiction (independent of Phase-4 ops)
    for label, frontier in states:
        rescan = frontier.get("final_rescan_new_groups_above_floor")
        if (
            frontier.get("descent_converged") is True
            and isinstance(rescan, int)
            and not isinstance(rescan, bool)
            and rescan > 0
        ):
            errors.append(
                f"{label}: descent_converged is true but "
                f"final_rescan_new_groups_above_floor is {rescan} (> 0) — a converged "
                "frontier must have a dry terminal re-scan; this contradiction means the "
                "descent was declared complete while shareable reuse remained one level down"
            )

    # The integrated schema-2 scanner inspects the complete nested definition
    # graph once. Recompute its coverage evidence instead of requiring a legacy
    # iterative frontier or pretending that a terminal rescan was performed.
    recursive_complete = False
    if manifests and all(m.get("schema_version") == 2 for m in manifests):
        import importlib.util
        path = Path(__file__).resolve().parents[2] / "usd-structure-assessment/scripts/structure_assessment.py"
        spec = importlib.util.spec_from_file_location("report_discovery_evidence", path)
        discovery = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(discovery)
        checks = [discovery.check_discovery_evidence(m) for m in manifests]
        recursive_complete = (report.get("target_coverage", {}).get("complete") is True
                              and all(check["passed"] for check in checks))
        if not recursive_complete:
            errors.append("recursive assessment coverage incomplete: " + str(
                [check["errors"] for check in checks if not check["passed"]]
                or "target coverage is incomplete"))
        return errors

    # FF1 — geometry op ran on a dedup/instancing run with no recorded convergence
    if dcc.did_dedup(report, manifests) and dcc.did_phase4_geom(report):
        if not recursive_complete and not any(dcc.is_converged_or_waived(frontier) for _, frontier in states):
            errors.append(
                "descent-convergence gate: a Phase-4 geometry op (decimation / "
                "within-prototype merge / primitive-fit / quantization / topology edit) "
                "ran on a dedup/instancing run, but no converged de-duplication frontier "
                "is recorded (no frontier.descent_converged == true with a dry terminal "
                "re-scan, and no recorded waiver). Record convergence in the "
                "apply-restructure manifest frontier block, or in report.structural_summary "
                "(frontier_descent_converged + frontier_final_rescan_new_groups_above_floor "
                "== 0), or set an explicit frontier.convergence_waived_by_user with a reason. "
                "A low/coarse prototype count is the SIGNAL TO DESCEND, not to merge: "
                "running geometry ops first fuses or coarsens geometry that further sharing "
                "would have collapsed (premature-merge inflation)."
            )
    return errors


def validate_completed_report(report: Any, manifests: list[Any] | None = None,
                              validator_concepts: Path | None = None,
                              schema: dict | None = None) -> list[str]:
    """Apply the same complete report contract to API and CLI results.

    Callers load referenced manifests before invoking this function; the CLI
    additionally reports failures to load its recorded manifest paths.
    """
    manifests = manifests or []
    errors = validate_report(report, schema)
    for index, manifest in enumerate(manifests):
        errors.extend(f"manifest[{index}]: {message}"
                      for message in validate_manifest_structure(manifest))
    errors.extend(reconcile_target_coverage(report, manifests))
    errors.extend(validate_footprint(report))
    errors.extend(validate_preservation(report))
    errors.extend(validate_safety_gate(report, validator_concepts))
    errors.extend(validate_score_arithmetic(report))
    errors.extend(validate_descent_convergence(report, manifests))
    return errors


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("report", type=Path, help="Path to the report JSON to validate.")
    parser.add_argument("--schema", type=Path, default=DEFAULT_SCHEMA)
    parser.add_argument(
        "--manifest",
        type=Path,
        action="append",
        default=[],
        help="apply-restructure manifest(s) to reconcile Phase-4 coverage against; "
        "repeat once per iteration so the union is checked. Manifests recorded in "
        "the report's target_coverage.source_manifests[] are loaded automatically "
        "and merged with these.",
    )
    parser.add_argument(
        "--validator-concepts",
        type=Path,
        default=None,
        help="Path to validator-concepts.json. Defaults to the sibling "
        "usd-validation-runner reference; the safety-gate check reads it to learn "
        "which concepts carry role 'safety_gate'.",
    )
    args = parser.parse_args()

    report = json.loads(args.report.read_text(encoding="utf-8"))
    schema = json.loads(args.schema.read_text(encoding="utf-8"))
    errors: list[str] = []

    labeled: list[tuple[str, Any]] = []
    for manifest_path in args.manifest:
        labeled.append((manifest_path.name, json.loads(manifest_path.read_text(encoding="utf-8"))))
    recorded, load_errors = load_recorded_manifests(report, args.report.resolve().parent)
    errors.extend(load_errors)
    labeled.extend(recorded)

    manifests = [manifest for _, manifest in labeled]
    errors.extend(validate_completed_report(report, manifests, args.validator_concepts, schema))

    if errors:
        print(f"{args.report}: INVALID ({len(errors)} error(s))")
        for error in errors:
            print(f"  {error}")
        return 1
    print(f"{args.report}: OK")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())