#!/usr/bin/env python3

"""Validate committed AVBD evidence packets against the shared schema.

Enforces PLAN-091 WP-091.1: AVBD packets at the current schema version

must machine-record the resolved solver configuration and the rigid-contact

selection source that actually ran (``resolved_solver_identity``). Packets

committed before the identity contract stay readable through a legacy

allowlist, but new packet files must be written at the current schema version.

The field contract lives in ``scripts/avbd_packet_schema.py``.

"""

from __future__ import annotations

import argparse

import hashlib

import json

import math

import re

import struct

import sys

import uuid

from collections.abc import Iterator, Mapping

from dataclasses import dataclass, field

from datetime import datetime

from fractions import Fraction

from pathlib import Path

from typing import Any

SCRIPT_DIR = Path(__file__).resolve().parent

if str(SCRIPT_DIR) not in sys.path:

sys.path.insert(0, str(SCRIPT_DIR))

from avbd_packet_schema import ( # noqa: E402

AVBD_PACKET_SCHEMA_VERSION,

MULTIBODY_IDENTITY_MIN_SCHEMA_VERSION,

PAPER_PACKET_SOURCE_PATHS,

PLAN104_CLAIMS_KEY,

PLAN104_CLAIMS_MIN_SCHEMA_VERSION,

SOLVER_CONFIGURATION_MIN_SCHEMA_VERSION,

SOURCE_PROVENANCE_ALGORITHM,

evidence_definition_matches,

packet_schema_version_errors,

plan104_claims_errors,

resolved_solver_identity_errors,

stable_counter_stddev_is_noise,

)

__all__ = [

"AVBD_PACKET_SCHEMA_VERSION",

"PacketValidationContext",

"packet_errors",

]

from capture_source_provenance import ( # noqa: E402

CAPTURE_ARTIFACT_PROVENANCE_ALGORITHM,

CAPTURE_LOADER_ENVIRONMENT_PREFIXES,

CAPTURE_LOADER_POLICY_ALGORITHM,

CAPTURE_PNG_SEQUENCE_PROVENANCE_ALGORITHM,

CAPTURE_RUNTIME_IMAGE_INVENTORY_ALGORITHM,

CAPTURE_RUNTIME_PROVENANCE_ALGORITHM,

CAPTURE_SCREENSHOT_BINDING_ALGORITHM,

CAPTURE_SOURCE_PROVENANCE_ALGORITHM,

CAPTURE_VIDEO_CONTENT_CORRESPONDENCE_ALGORITHM,

CAPTURE_VIDEO_ENCODER,

CAPTURE_VIDEO_PROBE_ALGORITHM,

DART_LIBRARY_BUILD_IDENTITY_ALGORITHM,

CaptureSourceProvenanceError,

compute_capture_source_provenance,

)

from run_figure13_benchmark import ( # noqa: E402

BUILD_CONFIGURATION_ALGORITHM,

BUILD_CONFIGURATION_KEYS,

EVIDENCE_CMAKE_DEFINITIONS,

LOADER_ENVIRONMENT_PREFIXES,

LOADER_POLICY_ALGORITHM,

REQUIRED_RUNTIME_IMAGE_ROLES,

RUNTIME_IMAGE_INVENTORY_ALGORITHM,

)

REPO_ROOT = SCRIPT_DIR.parent

DEFAULT_PACKET_DIR = REPO_ROOT / "docs" / "plans" / "104-vertex-block-descent-solver"

PACKET_GLOB = "avbd-*-packet.json"

LINKED_PACKET_KEYS = ("linked_avbd_evidence", "linked_avbd_vbd_evidence")

BENCHMARK_SOURCE_PATH = Path("tests/benchmark/simulation/bm_avbd_rigid_fixed_joint.cpp")

FIGURE13_BENCHMARK_FILTER = (

"^BM_(Avbd|Vbd|SequentialImpulse)PaperBreakableWallStep/iterations:120$"

)

FIGURE13_BENCHMARK_RUN_SCHEMA = "dart.figure13_benchmark_run/v3"

BENCHMARK_BUILD_IDENTITY_ALGORITHM = "sha256-canonical-compiled-benchmark-identity-v3"

BENCHMARK_SOURCE_PROVENANCE_ALGORITHM = (

"sha256-canonical-benchmark-source-and-executable-v3"

)

SEMANTIC_CLAIM_ASSESSMENTS = {

"capture_checkpoint_semantics": "supported",

"cuda_parity": "not_proven",

"exact_unpublished_source_scene": "not_proven",

"full_interval_video_semantics": "supported",

"other_figure13_solver_rows": "not_proven",

"paper_figure_visual_agreement": "supported",

"published_timing_parity": "not_proven",

"text_oracle_agreement": "supported",

"view_report_agreement": "supported",

"xpbd_parity": "not_proven",

}

SEMANTIC_TERMINAL_BEHAVIOR = {

"avbd-paper-breakable-wall-packet.json": "retained_damaged_wall",

"avbd-paper-vbd-comparison-packet.json": "bent_retained_wall",

"avbd-paper-sequential-impulse-comparison-packet.json": "collapsed_wall",

}

# The public AVBD reconstruction under the immutable paper profile keeps the

# wall standing with three small joint-break clusters and no displaced bricks;

# Figure 13(d) shows the wall broken open with flying bricks, so visual

# agreement with the paper figure is not proven for that row.

SEMANTIC_CLAIM_ASSESSMENTS_BY_TERMINAL_BEHAVIOR = {

"retained_damaged_wall": {

**SEMANTIC_CLAIM_ASSESSMENTS,

"paper_figure_visual_agreement": "not_proven",

},

"bent_retained_wall": dict(SEMANTIC_CLAIM_ASSESSMENTS),

"collapsed_wall": dict(SEMANTIC_CLAIM_ASSESSMENTS),

}

SEMANTIC_STRUCTURED_OBSERVATIONS = {

"retained_damaged_wall": {

"checkpoint_sequence": [

"three_localized_joint_break_clusters_at_frame_60",

"retained_damaged_wall_at_frame_120",

"retained_damaged_wall_at_frame_600",

],

"paper_figure_relationship": "retained_wall_without_the_paper_fracture",

"temporal_behavior": "retained_damaged_wall",

"text_oracle_relationship": "agrees",

"view_report_relationship": "agrees",

},

"bent_retained_wall": {

"checkpoint_sequence": [

"distributed_bend_at_frame_18",

"bent_retained_wall_at_frame_120",

"bent_retained_wall_at_frame_600",

],

"paper_figure_relationship": "qualitative_agreement",

"temporal_behavior": "bent_retained_wall",

"text_oracle_relationship": "agrees",

"view_report_relationship": "agrees",

},

"collapsed_wall": {

"checkpoint_sequence": [

"localized_three_region_fracture_at_frame_14",

"collapsed_wall_at_frame_120",

"collapsed_wall_at_frame_600",

],

"paper_figure_relationship": "qualitative_agreement",

"temporal_behavior": "collapsed_wall",

"text_oracle_relationship": "agrees",

"view_report_relationship": "agrees",

},

}

PAPER_CAPTURE_ROLES = {

"avbd-paper-breakable-wall-packet.json": ("impact", "outcome"),

"avbd-paper-vbd-comparison-packet.json": ("bend", "retention"),

"avbd-paper-sequential-impulse-comparison-packet.json": (

"fracture",

"collapse",

),

}

PAPER_LONG_HORIZON_ROLE = "long_horizon"

PAPER_LONG_HORIZON_FRAME = 600

# Checker-owned copies of the Figure 13 outcome oracles. A packet grades its

# own physical outcome against the oracle it ships, so the thresholds have to

# be pinned here as well: without an independent copy a regenerated packet

# could relax `minimum_broken_joints` to 0 and still report a passing

# `physical_outcome_valid`. These must stay byte-identical to the writers'

# `OUTCOME_ORACLE` constants, which

# tests/test_check_avbd_packets.py::test_paper_outcome_oracles_match_writer_constants

# proves on every run.

_BREAKABLE_WALL_OUTCOME_ORACLE: dict[str, object] = {

"evaluation_frame": 120,

"expected_broken_joint_ids_sha256": (

"e746389411f654ea64f2836db35c704443b2dac09186fc73d4a9341a18890fab"

),

"impact_band_radius": 0.85,

"impact_damage_displacement_threshold": 0.1,

"joint_evidence_frames": [60, 120, 600],

"maximum_broken_joints": 60,

"maximum_broken_joints_outside_impact_regions": 21,

"maximum_unbroken_joint_angular_residual_radians": 0.002,

"maximum_unbroken_joint_linear_residual": 0.002,

"minimum_broken_joints": 30,

"minimum_broken_joints_per_impact_region": 4,

"minimum_outside_retained_fraction": 0.95,

"minimum_total_retained_fraction": 0.95,

"minimum_unbroken_joints": 650,

"outside_radius": 1.15,

"retained_displacement_threshold": 0.5,

}

_VBD_COMPARISON_OUTCOME_ORACLE: dict[str, object] = {

"bent_brick_displacement_threshold": 0.05,

"evaluation_frame": 18,

"impact_band_radius": 0.85,

"impact_damage_displacement_threshold": 0.1,

"joint_evidence_frames": [18, 120, 600],

"maximum_broken_joints": 0,

"maximum_unbroken_joint_angular_residual_radians": 0.02,

"maximum_unbroken_joint_linear_residual": 0.025,

"minimum_bent_bricks": 100,

"minimum_maximum_wall_normal_displacement": 0.1,

"minimum_rms_wall_normal_displacement": 0.05,

"minimum_total_retained_fraction": 0.99,

"minimum_unbroken_joints": 712,

"outside_radius": 1.15,

"retained_displacement_threshold": 0.5,

"retention_evaluation_frame": 120,

}

_SEQUENTIAL_IMPULSE_OUTCOME_ORACLE: dict[str, object] = {

"collapse_evaluation_frame": 120,

"evaluation_frame": 14,

"expected_broken_joint_ids_sha256": (

"c85184879b1b9036ff582731031fc49c56b4149e93ded45780b06352bd94d61d"

),

"expected_broken_joints": 5,

"expected_outside_impact_unbroken_joint_count": 484,

"impact_band_radius": 0.85,

"impact_damage_displacement_threshold": 0.1,

"joint_evidence_frames": [14, 120, 600],

"maximum_collapse_outside_retained_fraction": 0.35,

"maximum_collapse_total_retained_fraction": 0.25,

"maximum_final_broken_joints": 30,

"maximum_initial_broken_joints": 20,

"maximum_initial_outside_joint_angular_residual_radians": 0.13,

"maximum_initial_outside_joint_linear_residual": 0.04,

"minimum_collapse_outside_joint_maximum_angular_residual_radians": 0.6,

"minimum_collapse_outside_joint_maximum_linear_residual": 0.05,

"minimum_collapse_outside_joint_rms_angular_residual_radians": 0.18,

"minimum_collapse_outside_joint_rms_linear_residual": 0.015,

"minimum_collapse_wall_normal_displacement": 2.0,

"minimum_displaced_bricks_per_impact_band": 10,

"minimum_final_broken_joints": 3,

"minimum_final_unbroken_joints": 680,

"minimum_initial_broken_joints": 3,

"minimum_initial_broken_joints_per_impact_region": 1,

"minimum_initial_total_retained_fraction": 0.95,

"minimum_initial_unbroken_joints": 690,

"outside_radius": 1.15,

"retained_displacement_threshold": 0.5,

}

PAPER_OUTCOME_ORACLES: dict[str, dict[str, dict[str, object]]] = {

packet_name: {

checkpoint: oracle for checkpoint in (*roles, PAPER_LONG_HORIZON_ROLE)

}

for packet_name, roles, oracle in (

(

"avbd-paper-breakable-wall-packet.json",

PAPER_CAPTURE_ROLES["avbd-paper-breakable-wall-packet.json"],

_BREAKABLE_WALL_OUTCOME_ORACLE,

),

(

"avbd-paper-vbd-comparison-packet.json",

PAPER_CAPTURE_ROLES["avbd-paper-vbd-comparison-packet.json"],

_VBD_COMPARISON_OUTCOME_ORACLE,

),

(

"avbd-paper-sequential-impulse-comparison-packet.json",

PAPER_CAPTURE_ROLES["avbd-paper-sequential-impulse-comparison-packet.json"],

_SEQUENTIAL_IMPULSE_OUTCOME_ORACLE,

),

)

}

PAPER_REQUIRED_MEDIAN_RATIO_KEYS = {

"avbd-paper-vbd-comparison-packet.json": frozenset(

{"vbd_to_avbd_median_cpu_cost_ratio"}

),

"avbd-paper-sequential-impulse-comparison-packet.json": frozenset(

{

"sequential_impulse_to_avbd_median_cpu_cost_ratio",

"sequential_impulse_to_vbd_median_cpu_cost_ratio",

}

),

}

PAPER_FIGURE13_SPECS: dict[str, dict[str, Any]] = {

"avbd-paper-breakable-wall-packet.json": {

"packet": "avbd_paper_breakable_wall",

"scene": "avbd_paper_breakable_wall",

"identity_solver": "avbd",

"capture_solver": "avbd",

"selection": "world_solver_family",

"public_solver": "AVBD",

"scene_solver": "public_avbd",

"captures": {

"impact": {

"review_role": "impact_frame_60",

"frame": 60,

"checkpoint": "outcome",

"evaluated": False,

"status": "pre-evaluation",

"thresholds_pass": False,

"evaluation_key": "evaluation_frame",

"evaluation_frame": 120,

"threshold_checks": (

"finite_state",

"fracture_activated",

"fracture_count_bounded",

"fracture_identity_matches",

"fracture_in_three_impact_regions",

"outside_breaks_bounded",

"outside_wall_retained",

"retained_joint_rows_satisfied",

"total_wall_retained",

),

},

"outcome": {

"review_role": "outcome_frame_120",

"frame": 120,

"checkpoint": "outcome",

"evaluated": True,

"status": "pass",

"thresholds_pass": True,

"evaluation_key": "evaluation_frame",

"evaluation_frame": 120,

"threshold_checks": (

"finite_state",

"fracture_activated",

"fracture_count_bounded",

"fracture_identity_matches",

"fracture_in_three_impact_regions",

"outside_breaks_bounded",

"outside_wall_retained",

"retained_joint_rows_satisfied",

"total_wall_retained",

),

},

"long_horizon": {

"frame": PAPER_LONG_HORIZON_FRAME,

"long_horizon": True,

"review_role": "long_horizon_frame_600",

"checkpoint": "outcome",

"evaluated": True,

"status": "pass",

"thresholds_pass": True,

"evaluation_key": "evaluation_frame",

"evaluation_frame": 120,

"threshold_checks": (

"finite_state",

"fracture_activated",

"fracture_count_bounded",

"fracture_identity_matches",

"fracture_in_three_impact_regions",

"outside_breaks_bounded",

"outside_wall_retained",

"retained_joint_rows_satisfied",

"total_wall_retained",

),

},

},

"benchmark_methods": {

"avbd": {

"location": "root",

"benchmark": "BM_AvbdPaperBreakableWallStep",

"runtime_solver": "avbd",

"family_index": None,

}

},

},

"avbd-paper-vbd-comparison-packet.json": {

"packet": "avbd_paper_vbd_comparison",

"scene": "vbd_paper_breakable_wall",

"identity_solver": "vbd",

"capture_solver": "vbd",

"selection": "world_solver_family",

"public_solver": "VBD",

"scene_solver": "public_vbd",

"captures": {

"bend": {

"review_role": "bend_frame_18",

"frame": 18,

"checkpoint": "bend",

"evaluated": True,

"status": "pass",

"thresholds_pass": True,

"evaluation_key": "evaluation_frame",

"evaluation_frame": 18,

"threshold_checks": (

"bend_is_spatially_resolved",

"finite_state",

"no_fracture",

"retained_joint_rows_satisfied",

"topology_retained",

"wall_bend_is_distributed",

"wall_bends",

),

},

"retention": {

"review_role": "retention_frame_120",

"frame": 120,

"checkpoint": "retention",

"evaluated": True,

"status": "pass",

"thresholds_pass": True,

"evaluation_key": "retention_evaluation_frame",

"evaluation_frame": 120,

"threshold_checks": (

"finite_state",

"no_fracture",

"retained_joint_rows_satisfied",

"topology_retained",

"wall_retained",

),

},

"long_horizon": {

"frame": PAPER_LONG_HORIZON_FRAME,

"long_horizon": True,

"review_role": "long_horizon_frame_600",

"checkpoint": "retention",

"evaluated": True,

"status": "pass",

"thresholds_pass": True,

"evaluation_key": "retention_evaluation_frame",

"evaluation_frame": 120,

"threshold_checks": (

"finite_state",

"no_fracture",

"retained_joint_rows_satisfied",

"topology_retained",

"wall_retained",

),

},

},

"benchmark_methods": {

"avbd": {

"location": "methods",

"benchmark": "BM_AvbdPaperBreakableWallStep",

"runtime_solver": "avbd",

"family_index": 0,

},

"vbd": {

"location": "methods",

"benchmark": "BM_VbdPaperBreakableWallStep",

"runtime_solver": "vbd",

"family_index": 1,

},

},

},

"avbd-paper-sequential-impulse-comparison-packet.json": {

"packet": "avbd_paper_sequential_impulse_comparison",

"scene": "sequential_impulse_paper_breakable_wall",

"identity_solver": "sequential_impulse",

"capture_solver": "sequential-impulse",

"selection": "contact_solver_method",

"public_solver": "SEQUENTIAL_IMPULSE",

"scene_solver": "public_sequential-impulse",

"captures": {

"fracture": {

"review_role": "fracture_frame_14",

"frame": 14,

"checkpoint": "fracture",

"evaluated": True,

"status": "pass",

"thresholds_pass": True,

"evaluation_key": "evaluation_frame",

"evaluation_frame": 14,

"threshold_checks": (

"finite_state",

"fracture_activated",

"initial_fracture_confined_to_impact_regions",

"initial_fracture_covers_three_impacts",

"initial_fracture_identity_matches",

"initial_retained_joint_rows_bounded",

"wall_initially_retained",

),

},

"collapse": {

"review_role": "collapse_frame_120",

"frame": 120,

"checkpoint": "collapse",

"evaluated": True,

"status": "pass",

"thresholds_pass": True,

"evaluation_key": "collapse_evaluation_frame",

"evaluation_frame": 120,

"threshold_checks": (

"damage_in_three_impact_bands",

"finite_state",

"fracture_identity_unchanged",

"initial_fracture_remains_visible",

"outside_wall_collapses",

"retained_rows_fail_outside_impacts",

"wall_collapses",

),

},

"long_horizon": {

"frame": PAPER_LONG_HORIZON_FRAME,

"long_horizon": True,

"review_role": "long_horizon_frame_600",

"checkpoint": "collapse",

"evaluated": True,

"status": "pass",

"thresholds_pass": True,

"evaluation_key": "collapse_evaluation_frame",

"evaluation_frame": 120,

"threshold_checks": (

"damage_in_three_impact_bands",

"finite_state",

"fracture_identity_unchanged",

"initial_fracture_remains_visible",

"outside_wall_collapses",

"retained_rows_fail_outside_impacts",

"wall_collapses",

),

},

},

"benchmark_methods": {

"sequential_impulse": {

"location": "method",

"benchmark": "BM_SequentialImpulsePaperBreakableWallStep",

"runtime_solver": "sequential-impulse",

"family_index": 2,

}

},

},

}

_PAPER_FINGERPRINT_HEX_LENGTH = 16

_PAPER_TRAJECTORY_FRAMES = 120

_PAPER_RIGID_CONSTRAINT_ITERATIONS = 20

_PAPER_BENCHMARK_REPETITIONS = 5

_PAPER_BENCHMARK_ITERATIONS = 5

_PAPER_CAPTURE_VIDEO_FPS = 60

_PAPER_SCENE_COUNTERS = {

"breakable_joints": 712,

"collision_shapes": 256,

"impacting_balls": 3,

"rigid_bodies": 256,

"rigid_body_joints": 712,

}

def _paper_requires_long_horizon(packet: Mapping[str, object]) -> bool:

version = packet.get("schema_version")

return (

isinstance(version, int)

and not isinstance(version, bool)

and version >= SOLVER_CONFIGURATION_MIN_SCHEMA_VERSION

)

def _paper_capture_roles(

packet: Mapping[str, object], packet_name: str

) -> tuple[str, ...] | None:

roles = PAPER_CAPTURE_ROLES.get(packet_name)

if roles is None:

return None

if _paper_requires_long_horizon(packet):

return (*roles, PAPER_LONG_HORIZON_ROLE)

return roles

HISTORICAL_HIGH_RATIO_BOUNDARIES = {

"avbd-articulated-high-ratio-chain-packet.json": {

"supported_scope": (

"historical_variational_multibody_capture_hash_and_cpu_metadata"

),

"visual_boundary": True,

},

"avbd-paper-scale-high-ratio-chain-packet.json": {

"supported_scope": (

"historical_variational_multibody_capture_hash_and_cpu_metadata"

),

"visual_boundary": True,

},

"avbd-paper-scale-high-ratio-iteration-sweep-packet.json": {

"supported_scope": ("historical_variational_multibody_cpu_metadata_and_plot"),

"visual_boundary": False,

},

}

# Packets committed before the resolved-solver-identity contract

# (WP-091.1). They remain readable at schema_version 1; their

# sequential-impulse contact rows are relabeled in prose instead of

# being rewritten. Do not add new packets here: new packet files must use the

# current AVBD_PACKET_SCHEMA_VERSION with a recorded identity.

LEGACY_IDENTITY_EXEMPT_PACKETS = frozenset(

{

"avbd-articulated-breakable-joint-packet.json",

"avbd-articulated-breakable-motor-packet.json",

"avbd-articulated-fixed-pair-breakable-joint-packet.json",

"avbd-articulated-high-ratio-chain-packet.json",

"avbd-articulated-prismatic-motor-packet.json",

"avbd-articulated-prismatic-pair-breakable-motor-packet.json",

"avbd-articulated-revolute-motor-packet.json",

"avbd-articulated-spherical-breakable-joint-packet.json",

"avbd-articulated-spherical-pair-breakable-joint-packet.json",

"avbd-articulated-world-prismatic-breakable-motor-packet.json",

"avbd-articulated-world-revolute-breakable-motor-packet.json",

"avbd-demo2d-cards-packet.json",

"avbd-demo2d-dynamic-friction-packet.json",

"avbd-demo2d-fracture-packet.json",

"avbd-demo2d-ground-packet.json",

"avbd-demo2d-hanging-rope-packet.json",

"avbd-demo2d-heavy-rope-packet.json",

"avbd-demo2d-joint-grid-packet.json",

"avbd-demo2d-motor-packet.json",

"avbd-demo2d-net-packet.json",

"avbd-demo2d-pyramid-packet.json",

"avbd-demo2d-rod-packet.json",

"avbd-demo2d-rope-packet.json",

"avbd-demo2d-soft-body-packet.json",

"avbd-demo2d-spring-packet.json",

"avbd-demo2d-spring-ratio-packet.json",

"avbd-demo2d-stack-packet.json",

"avbd-demo2d-stack-ratio-packet.json",

"avbd-demo2d-static-friction-packet.json",

"avbd-demo3d-breakable-packet.json",

"avbd-demo3d-bridge-packet.json",

"avbd-demo3d-dynamic-friction-packet.json",

"avbd-demo3d-ground-packet.json",

"avbd-demo3d-heavy-rope-packet.json",

"avbd-demo3d-pyramid-packet.json",

"avbd-demo3d-rope-packet.json",

"avbd-demo3d-soft-body-packet.json",

"avbd-demo3d-spring-packet.json",

"avbd-demo3d-spring-ratio-packet.json",

"avbd-demo3d-stack-packet.json",

"avbd-demo3d-stack-ratio-packet.json",

"avbd-demo3d-static-friction-packet.json",

"avbd-empty-baseline-packet.json",

"avbd-paper-scale-high-ratio-chain-packet.json",

"avbd-rigid-breakable-joint-packet.json",

"avbd-rigid-prismatic-motor-packet.json",

"avbd-rigid-revolute-motor-packet.json",

"avbd-rigid-spherical-breakable-joint-packet.json",

}

)

# These schema-version 3 packets were committed before schema version 4 moved

# public hard pair rows from the private AVBD compatibility projection into the

# Sequential Impulse family. Their writers now emit the current schema when

# regenerated. Keep only these exact historical names readable; a new filename

# must use AVBD_PACKET_SCHEMA_VERSION so it cannot claim the retired v3 solver

# identity contract. A filename leaves this set the moment its packet is

# regenerated at the current schema (the breakable scale packets left at

# schema 6), so a later downgrade of current evidence is rejected outright.

LEGACY_PRE_SI_PAIR_ROW_PACKETS = frozenset(

{

"avbd-articulated-compliant-fracture-packet.json",

"avbd-articulated-compliant-joints-packet.json",

"avbd-articulated-compliant-motors-packet.json",

"avbd-friction-coefficient-sweep-packet.json",

"avbd-paper-scale-high-ratio-iteration-sweep-packet.json",

}

)

# The schema-version 5 Figure 13 packets predate the solver-configuration

# fingerprint, multibody identity, and numeric Table 2 binding introduced by

# schema version 6. All three were regenerated at schema 6, so no committed

# packet remains at version 5 and that contract is retired: a Figure 13

# filename at schema 5 is rejected like any other stale version.

# Pin each historical filename to the one legacy version it was committed

# with. A legacy filename may move directly to the current schema when its

# packet is regenerated; once it has, it leaves this map so the current

# evidence cannot be downgraded back to a readable historical contract.

LEGACY_PACKET_SCHEMA_VERSIONS = {

**{name: 1 for name in LEGACY_IDENTITY_EXEMPT_PACKETS},

**{name: 3 for name in LEGACY_PRE_SI_PAIR_ROW_PACKETS},

}

LEGACY_SCHEMA_EXEMPT_PACKETS = frozenset(LEGACY_PACKET_SCHEMA_VERSIONS)

# Identity-free schema-v1 packets and the five schema-v3 packets whose

# historical labels did not identify the runtime solver remain readable only

# as explicitly bounded historical artifacts. The boundary lives in the

# packet, not just this allowlist, so downstream readers cannot accidentally

# promote a bare legacy skeleton into current AVBD evidence.

LEGACY_NON_EVIDENCE_BOUNDARY_SCOPES = {

**{

name: "historical_artifact_or_topology_metadata_only"

for name in LEGACY_IDENTITY_EXEMPT_PACKETS

},

"avbd-articulated-high-ratio-chain-packet.json": (

"historical_variational_multibody_capture_hash_and_cpu_metadata"

),

"avbd-paper-scale-high-ratio-chain-packet.json": (

"historical_variational_multibody_capture_hash_and_cpu_metadata"

),

"avbd-articulated-compliant-fracture-packet.json": (

"historical_variational_multibody_artifact_and_cpu_metadata_only"

),

"avbd-articulated-compliant-joints-packet.json": (

"historical_variational_multibody_artifact_and_cpu_metadata_only"

),

"avbd-articulated-compliant-motors-packet.json": (

"historical_variational_multibody_artifact_and_cpu_metadata_only"

),

"avbd-friction-coefficient-sweep-packet.json": (

"historical_sequential_impulse_cpu_and_visual_metadata_only"

),

"avbd-paper-scale-high-ratio-iteration-sweep-packet.json": (

"historical_variational_multibody_cpu_metadata_and_plot"

),

}

@dataclass

class PacketLoadResult:

packet: dict[str, object] | None

errors: tuple[str, ...]

payload: bytes | None

sha256: str | None

@dataclass

class PacketValidationContext:

"""Shared parse and validation state for one AVBD packet batch."""

packet_dir: Path = field(default_factory=lambda: DEFAULT_PACKET_DIR.resolve())

in_progress: set[Path] = field(default_factory=set)

loaded: dict[Path, PacketLoadResult] = field(default_factory=dict)

artifact_errors: dict[Path, tuple[str, ...]] = field(default_factory=dict)

initialization_error: str | None = field(default=None, init=False)

def __post_init__(self) -> None:

try:

self.packet_dir = self.packet_dir.resolve()

except (OSError, RuntimeError, ValueError, UnicodeError) as exc:

self.initialization_error = f"packet directory cannot be resolved ({exc})"

def parse_args(argv: list[str]) -> argparse.Namespace:

parser = argparse.ArgumentParser(description=__doc__)

parser.add_argument(

"--packet",

action="append",

type=Path,

default=None,

help="Explicit packet file to validate (repeatable); defaults to "

"every avbd-*-packet.json under --packet-dir.",

)

parser.add_argument(

"--packet-dir",

type=Path,

default=DEFAULT_PACKET_DIR,

help="Directory scanned for avbd-*-packet.json files.",

)

parser.add_argument(

"--stale-source",

choices=("error", "report"),

default="error",

help="How to treat sealed evidence whose recorded source state no "

"longer matches the working tree. 'error' (default) fails closed, "

"which is the bar for any parity or performance claim; 'report' "

"prints the stale seals as advisories so repository-wide lint does "

"not fail on unrelated source or dependency changes. Structural "

"and hash defects inside a packet always fail.",

)

return parser.parse_args(argv)

# Findings that only say the sealed evidence predates the current working tree.

# They never describe a corrupt or self-inconsistent packet.

STALE_SOURCE_PATTERNS = (

re.compile(r"\.source_provenance\.digest does not match current source state$"),

re.compile(r"benchmark source capture digest does not match current source state$"),

re.compile(

r"benchmark source hash does not match current benchmark translation unit$"

),

)

# The packet-level digest is recomputed from the listed files, so it is a stale

# seal only when a listed file also drifted; on its own it is a mutated packet.

# A file list that departs from the canonical paper-packet source contract is

# a structural defect and always stays a hard error.

PACKET_DIGEST_PATTERN = re.compile(

r": source_provenance\.digest does not match current listed source contents$"

)

PACKET_FILE_DRIFT_PATTERN = re.compile(

r"source_provenance\.files\[\d+\]\.(sha256 drifted for |path must be a regular file: )"

)

# Capture metadata (file count, roots, ignored paths) legitimately changes only

# together with the source digest; a lone mismatch is an edited packet.

CAPTURE_METADATA_PATTERN = re.compile(

r"^(?P<capture>.*\.source_provenance)\.(file_count|ignored_paths|roots) does not "

r"match current source state$"

)

CAPTURE_DIGEST_PATTERN = re.compile(

r"^(?P<capture>.*\.source_provenance)\.digest does not match current source state$"

)

def split_stale_source_findings(errors: list[str]) -> tuple[list[str], list[str]]:

"""Split validator output into hard errors and stale-seal advisories."""

# A listed source file that really changed produces both a per-file hash

# mismatch and a packet-digest mismatch (the digest is recomputed from the

# current contents). Either finding alone is an edited packet, not drift.

sha_drift_packets = {

error.split(": ", 1)[0]

for error in errors

if PACKET_FILE_DRIFT_PATTERN.search(error)

}

digest_mismatch_packets = {

error.split(": ", 1)[0]

for error in errors

if PACKET_DIGEST_PATTERN.search(error)

}

drifted_packets = sha_drift_packets & digest_mismatch_packets

drifted_captures = {

match.group("capture")

for error in errors

if (match := CAPTURE_DIGEST_PATTERN.match(error))

}

hard: list[str] = []

stale: list[str] = []

for error in errors:

packet_name = error.split(": ", 1)[0]

metadata = CAPTURE_METADATA_PATTERN.match(error)

if any(pattern.search(error) for pattern in STALE_SOURCE_PATTERNS):

stale.append(error)

elif metadata and metadata.group("capture") in drifted_captures:

stale.append(error)

elif (

PACKET_DIGEST_PATTERN.search(error)

or PACKET_FILE_DRIFT_PATTERN.search(error)

) and packet_name in drifted_packets:

stale.append(error)

else:

hard.append(error)

return hard, stale

def _sha256(path: Path) -> str:

digest = hashlib.sha256()

with path.open("rb") as file:

for chunk in iter(lambda: file.read(1024 * 1024), b""):

digest.update(chunk)

return digest.hexdigest()

def _safe_relative_path(value: object) -> Path | None:

if not isinstance(value, str) or not value:

return None

if "\x00" in value:

return None

try:

value.encode("utf-8")

path = Path(value)

except OSError, RuntimeError, TypeError, ValueError, UnicodeError:

return None

if path.is_absolute() or ".." in path.parts or path.as_posix() != value:

return None

return path

def _reject_duplicate_json_keys(

pairs: list[tuple[str, object]],

) -> dict[str, object]:

result: dict[str, object] = {}

for key, value in pairs:

if key in result:

raise ValueError(f"duplicate object key {key!r}")

result[key] = value

return result

def _reject_nonstandard_json_constant(value: str) -> None:

raise ValueError(f"non-standard numeric constant {value!r}")

def _parse_finite_json_float(value: str) -> float:

parsed = float(value)

if not math.isfinite(parsed):

raise ValueError(f"non-finite JSON number {value!r}")

return parsed

def _strict_json_loads(payload: bytes) -> object:

text = payload.decode("utf-8")

return json.loads(

text,

object_pairs_hook=_reject_duplicate_json_keys,

parse_constant=_reject_nonstandard_json_constant,

parse_float=_parse_finite_json_float,

)

def _source_provenance_errors(

packet: Mapping[str, object], packet_name: str

) -> list[str]:

provenance = packet.get("source_provenance")

if provenance is None:

if packet_name in PAPER_CAPTURE_ROLES:

return [

f"{packet_name}: source_provenance must be an object for a "

"current paper packet"

]

# Every packet written at the current schema contract must name the

# source it was produced from, and so must any packet that closes a

# PLAN-104 row, whatever its version. Only packets still validated at

# their pinned legacy version keep their allowlisted provenance-free

# shape, because they predate the contract.

version = packet.get("schema_version")

current_schema = (

isinstance(version, int)

and not isinstance(version, bool)

and version >= AVBD_PACKET_SCHEMA_VERSION

)

if current_schema:

return [

f"{packet_name}: source_provenance must be an object for a "

f"packet written at schema_version {AVBD_PACKET_SCHEMA_VERSION}"

]

if PLAN104_CLAIMS_KEY in packet:

return [

f"{packet_name}: source_provenance must be an object for a "

f"packet that records {PLAN104_CLAIMS_KEY}"

]

return []

if not isinstance(provenance, Mapping):

return [f"{packet_name}: source_provenance must be an object"]

errors: list[str] = []

if provenance.get("algorithm") != SOURCE_PROVENANCE_ALGORITHM:

errors.append(

f"{packet_name}: source_provenance.algorithm must be "

f"{SOURCE_PROVENANCE_ALGORITHM!r}"

)

files = provenance.get("files")

if not isinstance(files, list) or not files:

return errors + [

f"{packet_name}: source_provenance.files must be a non-empty list"

]

required_paths = PAPER_PACKET_SOURCE_PATHS.get(packet_name)

actual_paths = [

entry.get("path") if isinstance(entry, Mapping) else None for entry in files

]

if required_paths is not None and actual_paths != list(required_paths):

errors.append(

f"{packet_name}: source_provenance.files paths must exactly match "

"the canonical ordered paper-packet source contract"

)

combined = hashlib.sha256()

seen: set[str] = set()

try:

repository_root = REPO_ROOT.resolve()

except (OSError, RuntimeError, ValueError, UnicodeError) as exc:

return errors + [f"{packet_name}: repository root cannot be resolved ({exc})"]

for index, entry in enumerate(files):

label = f"{packet_name}: source_provenance.files[{index}]"

if not isinstance(entry, Mapping):

errors.append(f"{label} must be an object")

continue

relative = _safe_relative_path(entry.get("path"))

if relative is None:

errors.append(f"{label}.path must be a safe repository-relative path")

continue

relative_text = relative.as_posix()

if relative_text in seen:

errors.append(f"{label}.path duplicates {relative_text!r}")

continue

seen.add(relative_text)

try:

lexical_source_path = REPO_ROOT / relative

if lexical_source_path.is_symlink():

errors.append(

f"{label}.path cannot be a symbolic link: {relative_text}"

)

continue

source_path = lexical_source_path.resolve()

except (OSError, RuntimeError, ValueError, UnicodeError) as exc:

errors.append(f"{label}.path cannot be resolved: {relative_text} ({exc})")

continue

try:

source_path.relative_to(repository_root)

except ValueError:

errors.append(

f"{label}.path resolves outside the repository: {relative_text}"

)

continue

try:

is_file = source_path.is_file()

except (OSError, ValueError, UnicodeError) as exc:

errors.append(f"{label}.path cannot be inspected: {relative_text} ({exc})")

continue

if not is_file:

errors.append(f"{label}.path must be a regular file: {relative_text}")

continue

try:

payload = source_path.read_bytes()

except (OSError, ValueError, UnicodeError) as exc:

errors.append(f"{label}.path cannot be read: {relative_text} ({exc})")

continue

current_hash = hashlib.sha256(payload).hexdigest()

if entry.get("sha256") != current_hash:

errors.append(

f"{label}.sha256 drifted for {relative_text}: expected "

f"{current_hash}"

)

try:

encoded_path = relative_text.encode("utf-8")

except UnicodeError as exc:

errors.append(f"{label}.path cannot be encoded as UTF-8 ({exc})")

continue

combined.update(struct.pack("<Q", len(encoded_path)))