Skip to content

Evidence documents

Benchmark, traced-allocation and scientific-validation outputs use the strict versioned models in hebog.validation.evidence, so measurements keep their provenance without implying that an exploratory run passed a gate. hebog.validation is repository tooling, not installed from the wheel; use it from a source checkout after uv sync --all-groups.

Every document has schema_version (currently 1) and a discriminating evidence_type; a stable run identifier and timezone-aware timestamp; exploratory or reviewed status; the dataset identifier, role, content checksum, (y, x) shape and workload class; the exact configuration checksum; and strict fields that reject unknown data.

Benchmark evidence

A benchmark document identifies the measured implementation by version and/or 40-character commit, container image digest where available and a checksum of the complete installed dependency inventory, with related Rapthor, LSMTool, Hebog or PyBDSF identities recorded separately; the environment has its own checksum so runs are never compared on display versions alone. Each repetition distinguishes warm-up from measured work and records wall and CPU seconds, peak resident memory, array-copy count and bytes, Dask task count, and transfer and spill bytes. Wall time, CPU time and peak RSS are required; optional instrumentation uses null only with a reason in unavailable_metrics, and zero always means a measured zero. A reviewed benchmark has at least one warm-up and five measured repetitions.

Resource records hold the executor kind, node/worker/thread topology, allocated cores, node memory, worker limits, reserved headroom and a storage identifier; aggregate worker limits must fit inside node memory after headroom. Reviewed multi-node evidence adds plane count, tile core and maximum halo, partition and graph task counts, scheduler overhead, worker occupancy, storage throughput, retries, stragglers and strong- and weak-scaling efficiency; separate documents per node count form the controlled scalability curve.

Scientific-comparison evidence

A scientific document identifies candidate and reference software separately and records a SHA-256 digest of each side's canonical product manifest plus the beam and match gate the comparison oracle used, so the report is bound to the exact catalogue, RMS and mask artifacts rather than to the input dataset alone. It embeds the complete product reports. Released PyBDSF and pinned master therefore produce separate documents even on the same dataset and candidate output.

Writing and loading evidence

from pathlib import Path

from hebog.validation.evidence import load_evidence, write_evidence

write_evidence(Path("benchmark-results/run.json"), evidence)
reloaded = load_evidence(Path("benchmark-results/run.json"))

The writer sorts keys, rejects non-finite values, appends a final newline and replaces the destination only after writing a temporary file. load_evidence accepts benchmark, scientific-comparison and traced-allocation documents. A traced-allocation document, which just traced-peak writes, records each repetition's traced process peak, the peak of the find_sources call alone and the import floor; the contribution guide says when it counts as reviewed. Raw evidence stays under the ignored benchmark-results/ directory or controlled external storage; commit only compact reviewed summaries and reproduction metadata. The models expose model_json_schema(), and a breaking change bumps the integer version without migration support before 1.0 (ADR-006).

Reference baselines

config/baselines/ holds the reviewed compact and representative benchmark documents for released PyBDSF and pinned master, the reference-product manifest that binds the frozen PyBDSF products, the master-versus-release scientific comparison, the one-tile overhead record (the strict model in hebog.validation.overhead) and three inventories without a model: phase-0-reference-environments.json (sanitized package inventories, runner and compiler hashes, verified checkouts and the explicit 5.0/3.0 profile), phase-0-representative-dataset.json and phase-0-starting-revisions.json. A unit test loads every typed record through its model and requires the directory to contain nothing else. The master-versus-release comparison is written by scripts/validation/compare_reference_products.py, which the equivalence lane reruns against the committed record, so a change to the comparison oracle that moves it fails there.

hebog.validation.evidence

Versioned machine-readable scientific and performance evidence.

Evidence documents distinguish exploratory observations from reviewed release evidence and distinguish unavailable instrumentation from measured zeroes. External product readers and benchmark runners create these records; loading this module never reads data or starts work.

TRACED_PEAK_TOLERANCE_BYTES = 2 ** 20 // 10 module-attribute

Largest spread of traced peaks that still counts as one peak.

A traced peak repeats to a few kilobytes rather than to the byte, because the strings, paths and metadata of one run allocate slightly differently in the next. A tenth of a mebibyte is the precision a peak is quoted at, about ten times the spread measured across repetitions, and two orders of magnitude below the image-sized arrays a scalability change moves.

BenchmarkEvidence

Bases: _EvidenceDocument

Versioned measurements for one exact implementation and environment.

Source code in src/hebog/validation/evidence.py 
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
class BenchmarkEvidence(_EvidenceDocument):
    """Versioned measurements for one exact implementation and environment."""

    evidence_type: Literal["benchmark"]
    subject: SoftwareIdentity
    related_software: tuple[SoftwareIdentity, ...] = ()
    environment_sha256: str = Field(pattern=_SHA256_PATTERN)
    resources: ResourceAllocation
    measurements: tuple[Measurement, ...] = Field(min_length=1)
    storage: StorageEvidence | None = None
    scalability: ScalabilityMetrics | None = None

    @model_validator(mode="after")
    def validate_benchmark_protocol(self) -> Self:
        """Enforce identity, ordering, and reviewed repetition requirements."""
        software_names = [
            self.subject.name,
            *(software.name for software in self.related_software),
        ]
        if len(set(software_names)) != len(software_names):
            raise ValueError("software identity names must be unique")

        repetition_indices = [
            measurement.repetition_index for measurement in self.measurements
        ]
        if len(set(repetition_indices)) != len(repetition_indices):
            raise ValueError("repetition indices must be unique")
        if repetition_indices != sorted(repetition_indices):
            raise ValueError(
                "measurements must be ordered by repetition index"
            )

        measured_repetition_seen = False
        for measurement in self.measurements:
            if measurement.warmup and measured_repetition_seen:
                raise ValueError(
                    "warm-up repetitions must precede measured repetitions"
                )
            measured_repetition_seen = (
                measured_repetition_seen or not measurement.warmup
            )

        if self.status is EvidenceStatus.REVIEWED:
            warmup_count = sum(
                measurement.warmup for measurement in self.measurements
            )
            measured_count = len(self.measurements) - warmup_count
            if warmup_count < 1:
                raise ValueError("reviewed evidence requires a warm-up")
            if measured_count < _MINIMUM_REVIEWED_REPETITIONS:
                raise ValueError(
                    "reviewed evidence requires five measured repetitions"
                )
            if self.resources.worker_nodes > 1 and self.scalability is None:
                raise ValueError(
                    "reviewed multi-node evidence requires scalability metrics"
                )
        return self

validate_benchmark_protocol()

Enforce identity, ordering, and reviewed repetition requirements.

Source code in src/hebog/validation/evidence.py 
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
@model_validator(mode="after")
def validate_benchmark_protocol(self) -> Self:
    """Enforce identity, ordering, and reviewed repetition requirements."""
    software_names = [
        self.subject.name,
        *(software.name for software in self.related_software),
    ]
    if len(set(software_names)) != len(software_names):
        raise ValueError("software identity names must be unique")

    repetition_indices = [
        measurement.repetition_index for measurement in self.measurements
    ]
    if len(set(repetition_indices)) != len(repetition_indices):
        raise ValueError("repetition indices must be unique")
    if repetition_indices != sorted(repetition_indices):
        raise ValueError(
            "measurements must be ordered by repetition index"
        )

    measured_repetition_seen = False
    for measurement in self.measurements:
        if measurement.warmup and measured_repetition_seen:
            raise ValueError(
                "warm-up repetitions must precede measured repetitions"
            )
        measured_repetition_seen = (
            measured_repetition_seen or not measurement.warmup
        )

    if self.status is EvidenceStatus.REVIEWED:
        warmup_count = sum(
            measurement.warmup for measurement in self.measurements
        )
        measured_count = len(self.measurements) - warmup_count
        if warmup_count < 1:
            raise ValueError("reviewed evidence requires a warm-up")
        if measured_count < _MINIMUM_REVIEWED_REPETITIONS:
            raise ValueError(
                "reviewed evidence requires five measured repetitions"
            )
        if self.resources.worker_nodes > 1 and self.scalability is None:
            raise ValueError(
                "reviewed multi-node evidence requires scalability metrics"
            )
    return self

DatasetIdentity

Bases: _EvidenceModel

Governed dataset and workload identity for evidence.

Source code in src/hebog/validation/evidence.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
class DatasetIdentity(_EvidenceModel):
    """Governed dataset and workload identity for evidence."""

    identifier: str = Field(pattern=r"^[a-z0-9]+(?:-[a-z0-9]+)*$")
    role: DatasetRole
    content_sha256: str = Field(pattern=_SHA256_PATTERN)
    shape_yx: tuple[int, int]
    workload_class: WorkloadClass

    @model_validator(mode="after")
    def validate_shape(self) -> Self:
        """Reject empty or negative logical image dimensions."""
        if any(dimension <= 0 for dimension in self.shape_yx):
            raise ValueError("shape_yx dimensions must be positive")
        return self

validate_shape()

Reject empty or negative logical image dimensions.

Source code in src/hebog/validation/evidence.py 
122
123
124
125
126
127
@model_validator(mode="after")
def validate_shape(self) -> Self:
    """Reject empty or negative logical image dimensions."""
    if any(dimension <= 0 for dimension in self.shape_yx):
        raise ValueError("shape_yx dimensions must be positive")
    return self

EvidenceStatus

Bases: str, Enum

Review status of an evidence document.

Source code in src/hebog/validation/evidence.py 
51
52
53
54
55
class EvidenceStatus(str, Enum):
    """Review status of an evidence document."""

    EXPLORATORY = "exploratory"
    REVIEWED = "reviewed"

ExecutorKind

Bases: str, Enum

Execution boundary used for a measured run.

Source code in src/hebog/validation/evidence.py 
66
67
68
69
70
71
72
class ExecutorKind(str, Enum):
    """Execution boundary used for a measured run."""

    SERIAL = "serial"
    LOCAL = "local"
    DASK = "dask"
    EXTERNAL = "external"

Measurement

Bases: _EvidenceModel

One warm-up or measured complete-run repetition.

Source code in src/hebog/validation/evidence.py 
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
class Measurement(_EvidenceModel):
    """One warm-up or measured complete-run repetition."""

    repetition_index: int = Field(ge=0)
    warmup: bool
    complete: RuntimeMetrics
    stages: tuple[StageMetrics, ...] = Field(min_length=1)

    @model_validator(mode="after")
    def validate_unique_stages(self) -> Self:
        """Prevent ambiguous repeated stage metrics in one repetition."""
        names = [stage.stage for stage in self.stages]
        if len(set(names)) != len(names):
            raise ValueError("stage names must be unique within a measurement")
        return self

validate_unique_stages()

Prevent ambiguous repeated stage metrics in one repetition.

Source code in src/hebog/validation/evidence.py 
274
275
276
277
278
279
280
@model_validator(mode="after")
def validate_unique_stages(self) -> Self:
    """Prevent ambiguous repeated stage metrics in one repetition."""
    names = [stage.stage for stage in self.stages]
    if len(set(names)) != len(names):
        raise ValueError("stage names must be unique within a measurement")
    return self

ResourceAllocation

Bases: _EvidenceModel

Admitted compute and storage topology for one measured run.

Source code in src/hebog/validation/evidence.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
class ResourceAllocation(_EvidenceModel):
    """Admitted compute and storage topology for one measured run."""

    executor: ExecutorKind
    worker_nodes: int = Field(ge=1)
    workers_per_node: int = Field(ge=1)
    threads_per_worker: int = Field(ge=1)
    allocated_cpu_cores: int = Field(ge=1)
    node_memory_bytes: int = Field(ge=1)
    worker_memory_limit_bytes: int = Field(ge=1)
    reserved_headroom_per_node_bytes: int = Field(ge=0)
    storage_identifier: str = Field(min_length=1)

    @model_validator(mode="after")
    def validate_memory_admission(self) -> Self:
        """Keep aggregate worker limits within admitted node memory."""
        available_memory = (
            self.node_memory_bytes - self.reserved_headroom_per_node_bytes
        )
        admitted_worker_memory = (
            self.workers_per_node * self.worker_memory_limit_bytes
        )
        if available_memory <= 0:
            raise ValueError("reserved headroom must leave usable node memory")
        if admitted_worker_memory > available_memory:
            raise ValueError(
                "aggregate worker memory exceeds node memory after headroom"
            )
        return self

validate_memory_admission()

Keep aggregate worker limits within admitted node memory.

Source code in src/hebog/validation/evidence.py 
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
@model_validator(mode="after")
def validate_memory_admission(self) -> Self:
    """Keep aggregate worker limits within admitted node memory."""
    available_memory = (
        self.node_memory_bytes - self.reserved_headroom_per_node_bytes
    )
    admitted_worker_memory = (
        self.workers_per_node * self.worker_memory_limit_bytes
    )
    if available_memory <= 0:
        raise ValueError("reserved headroom must leave usable node memory")
    if admitted_worker_memory > available_memory:
        raise ValueError(
            "aggregate worker memory exceeds node memory after headroom"
        )
    return self

RuntimeMetrics

Bases: _EvidenceModel

Required timing/resource metrics with explicit unavailable values.

Source code in src/hebog/validation/evidence.py 
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
class RuntimeMetrics(_EvidenceModel):
    """Required timing/resource metrics with explicit unavailable values."""

    wall_seconds: float = Field(ge=0, allow_inf_nan=False)
    cpu_seconds: float = Field(ge=0, allow_inf_nan=False)
    peak_rss_bytes: int = Field(ge=0)
    array_copy_count: int | None = Field(ge=0)
    array_copy_bytes: int | None = Field(ge=0)
    dask_task_count: int | None = Field(ge=0)
    transfer_bytes: int | None = Field(ge=0)
    spill_bytes: int | None = Field(ge=0)
    unavailable_metrics: tuple[UnavailableMetric, ...] = ()

    @model_validator(mode="after")
    def validate_unavailable_metrics(self) -> Self:
        """Require one reason for every unavailable optional instrument."""
        metric_values: dict[OptionalMetricName, int | None] = {
            "array_copy_count": self.array_copy_count,
            "array_copy_bytes": self.array_copy_bytes,
            "dask_task_count": self.dask_task_count,
            "transfer_bytes": self.transfer_bytes,
            "spill_bytes": self.spill_bytes,
        }
        reason_names = [item.metric for item in self.unavailable_metrics]
        if len(set(reason_names)) != len(reason_names):
            raise ValueError("unavailable metric names must be unique")
        unavailable_reasons = {
            item.metric: item.reason for item in self.unavailable_metrics
        }
        for metric_name, value in metric_values.items():
            reason = unavailable_reasons.get(metric_name)
            if value is None and reason is None:
                raise ValueError(
                    f"{metric_name} is unavailable without a reason"
                )
            if value is not None and reason is not None:
                raise ValueError(
                    f"{metric_name} has both a value and unavailable reason"
                )
        return self

validate_unavailable_metrics()

Require one reason for every unavailable optional instrument.

Source code in src/hebog/validation/evidence.py 
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
@model_validator(mode="after")
def validate_unavailable_metrics(self) -> Self:
    """Require one reason for every unavailable optional instrument."""
    metric_values: dict[OptionalMetricName, int | None] = {
        "array_copy_count": self.array_copy_count,
        "array_copy_bytes": self.array_copy_bytes,
        "dask_task_count": self.dask_task_count,
        "transfer_bytes": self.transfer_bytes,
        "spill_bytes": self.spill_bytes,
    }
    reason_names = [item.metric for item in self.unavailable_metrics]
    if len(set(reason_names)) != len(reason_names):
        raise ValueError("unavailable metric names must be unique")
    unavailable_reasons = {
        item.metric: item.reason for item in self.unavailable_metrics
    }
    for metric_name, value in metric_values.items():
        reason = unavailable_reasons.get(metric_name)
        if value is None and reason is None:
            raise ValueError(
                f"{metric_name} is unavailable without a reason"
            )
        if value is not None and reason is not None:
            raise ValueError(
                f"{metric_name} has both a value and unavailable reason"
            )
    return self

ScalabilityMetrics

Bases: _EvidenceModel

Topology and efficiency evidence required for multi-node runs.

Source code in src/hebog/validation/evidence.py 
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
class ScalabilityMetrics(_EvidenceModel):
    """Topology and efficiency evidence required for multi-node runs."""

    logical_plane_count: int = Field(ge=1)
    tile_core_shape_yx: tuple[int, int]
    maximum_halo_yx: tuple[int, int]
    partition_count: int = Field(ge=1)
    graph_task_count: int = Field(ge=0)
    scheduler_overhead_seconds: float = Field(ge=0, allow_inf_nan=False)
    worker_occupancy_fraction: float = Field(
        ge=0,
        le=1,
        allow_inf_nan=False,
    )
    storage_throughput_bytes_per_second: float = Field(
        ge=0,
        allow_inf_nan=False,
    )
    retry_count: int = Field(ge=0)
    straggler_count: int = Field(ge=0)
    strong_scaling_efficiency: float = Field(ge=0, allow_inf_nan=False)
    weak_scaling_efficiency: float = Field(ge=0, allow_inf_nan=False)

    @model_validator(mode="after")
    def validate_geometry(self) -> Self:
        """Require positive tile cores and non-negative halos."""
        if any(dimension <= 0 for dimension in self.tile_core_shape_yx):
            raise ValueError("tile core dimensions must be positive")
        if any(dimension < 0 for dimension in self.maximum_halo_yx):
            raise ValueError("halo dimensions cannot be negative")
        return self

validate_geometry()

Require positive tile cores and non-negative halos.

Source code in src/hebog/validation/evidence.py 
306
307
308
309
310
311
312
313
@model_validator(mode="after")
def validate_geometry(self) -> Self:
    """Require positive tile cores and non-negative halos."""
    if any(dimension <= 0 for dimension in self.tile_core_shape_yx):
        raise ValueError("tile core dimensions must be positive")
    if any(dimension < 0 for dimension in self.maximum_halo_yx):
        raise ValueError("halo dimensions cannot be negative")
    return self

ScientificComparisonEvidence

Bases: _EvidenceDocument

Provenance and reports for one candidate/reference comparison.

Source code in src/hebog/validation/evidence.py 
474
475
476
477
478
479
480
481
482
483
484
485
486
487
class ScientificComparisonEvidence(_EvidenceDocument):
    """Provenance and reports for one candidate/reference comparison."""

    evidence_type: Literal["scientific-comparison"]
    candidate: SoftwareIdentity
    reference: SoftwareIdentity
    candidate_product_manifest_sha256: str = Field(pattern=_SHA256_PATTERN)
    reference_product_manifest_sha256: str = Field(pattern=_SHA256_PATTERN)
    beam_fwhm_degrees: float = Field(gt=0, allow_inf_nan=False)
    maximum_separation_beams: float = Field(gt=0, allow_inf_nan=False)
    catalogue: CatalogueComparisonReport
    true_sky_rms: RmsComparisonReport
    flat_noise_rms: RmsComparisonReport
    mask: MaskComparisonReport

SoftwareIdentity

Bases: _EvidenceModel

Exact software and dependency identity used by one run.

Source code in src/hebog/validation/evidence.py 
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
class SoftwareIdentity(_EvidenceModel):
    """Exact software and dependency identity used by one run."""

    name: str = Field(pattern=r"^[a-z0-9]+(?:-[a-z0-9]+)*$")
    version: str | None = Field(default=None, min_length=1)
    commit_sha: str | None = Field(default=None, pattern=_COMMIT_PATTERN)
    source_tree_sha256: str | None = Field(
        default=None,
        pattern=_SHA256_PATTERN,
    )
    container_image_digest: str | None = Field(
        default=None,
        pattern=_CONTAINER_DIGEST_PATTERN,
    )
    dependency_inventory_sha256: str = Field(pattern=_SHA256_PATTERN)

    @model_validator(mode="after")
    def validate_revision(self) -> Self:
        """Require at least one independently useful revision identifier."""
        if (
            self.version is None
            and self.commit_sha is None
            and self.source_tree_sha256 is None
            and self.container_image_digest is None
        ):
            raise ValueError(
                "software identity requires version, commit_sha, "
                "source_tree_sha256, or container_image_digest"
            )
        return self

validate_revision()

Require at least one independently useful revision identifier.

Source code in src/hebog/validation/evidence.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
@model_validator(mode="after")
def validate_revision(self) -> Self:
    """Require at least one independently useful revision identifier."""
    if (
        self.version is None
        and self.commit_sha is None
        and self.source_tree_sha256 is None
        and self.container_image_digest is None
    ):
        raise ValueError(
            "software identity requires version, commit_sha, "
            "source_tree_sha256, or container_image_digest"
        )
    return self

StageMetrics

Bases: _EvidenceModel

Metrics attributed to one named scientific or integration stage.

Source code in src/hebog/validation/evidence.py 
259
260
261
262
263
class StageMetrics(_EvidenceModel):
    """Metrics attributed to one named scientific or integration stage."""

    stage: str = Field(pattern=r"^[a-z0-9]+(?:-[a-z0-9]+)*$")
    metrics: RuntimeMetrics

StorageEvidence

Bases: _EvidenceModel

Intermediate-store layout, policy, and observed footprint.

Source code in src/hebog/validation/evidence.py 
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
class StorageEvidence(_EvidenceModel):
    """Intermediate-store layout, policy, and observed footprint."""

    format_name: str = Field(min_length=1)
    library_name: str = Field(min_length=1)
    library_version: str = Field(min_length=1)
    backend_name: str = Field(min_length=1)
    chunk_shape_yx: tuple[int, int]
    shard_shape_yx: tuple[int, int] | None
    codec_pipeline: tuple[str, ...] = Field(min_length=1)
    fill_value: str = Field(min_length=1)
    missing_chunk_policy: Literal["error", "fill"]
    write_empty_chunks: bool
    object_count: int = Field(ge=1)
    stored_bytes: int = Field(ge=1)
    internal_concurrency: int = Field(ge=1)
    atomic_write_guarantee: str = Field(min_length=1)
    conditional_create: bool

    @model_validator(mode="after")
    def validate_storage_configuration(self) -> Self:
        """Require usable geometry and non-blank policy descriptions."""
        if any(dimension <= 0 for dimension in self.chunk_shape_yx):
            raise ValueError("storage chunk dimensions must be positive")
        if self.shard_shape_yx is not None and any(
            dimension <= 0 for dimension in self.shard_shape_yx
        ):
            raise ValueError("storage shard dimensions must be positive")
        descriptions = (
            self.format_name,
            self.library_name,
            self.library_version,
            self.backend_name,
            self.fill_value,
            self.atomic_write_guarantee,
            *self.codec_pipeline,
        )
        if any(not description.strip() for description in descriptions):
            raise ValueError("storage descriptions must not be blank")
        return self

validate_storage_configuration()

Require usable geometry and non-blank policy descriptions.

Source code in src/hebog/validation/evidence.py 
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
@model_validator(mode="after")
def validate_storage_configuration(self) -> Self:
    """Require usable geometry and non-blank policy descriptions."""
    if any(dimension <= 0 for dimension in self.chunk_shape_yx):
        raise ValueError("storage chunk dimensions must be positive")
    if self.shard_shape_yx is not None and any(
        dimension <= 0 for dimension in self.shard_shape_yx
    ):
        raise ValueError("storage shard dimensions must be positive")
    descriptions = (
        self.format_name,
        self.library_name,
        self.library_version,
        self.backend_name,
        self.fill_value,
        self.atomic_write_guarantee,
        *self.codec_pipeline,
    )
    if any(not description.strip() for description in descriptions):
        raise ValueError("storage descriptions must not be blank")
    return self

TracedAllocationEvidence

Bases: _EvidenceDocument

Deterministic traced-allocation peaks for one input and configuration.

tracemalloc counts the allocations the process itself makes, so its peak is reproducible for one input, one configuration and one implementation, which is why it, and not peak resident memory, gates a public envelope raise. Tracing roughly doubles wall time, so these documents record no timing that may be compared with benchmark evidence.

Source code in src/hebog/validation/evidence.py 
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
class TracedAllocationEvidence(_EvidenceDocument):
    """Deterministic traced-allocation peaks for one input and configuration.

    ``tracemalloc`` counts the allocations the process itself makes, so its
    peak is reproducible for one input, one configuration and one
    implementation, which is why it, and not peak resident memory, gates a
    public envelope raise. Tracing roughly doubles wall time, so these
    documents record no timing that may be compared with benchmark evidence.
    """

    evidence_type: Literal["traced-allocation"]
    subject: SoftwareIdentity
    environment_sha256: str = Field(pattern=_SHA256_PATTERN)
    resources: ResourceAllocation
    measurements: tuple[TracedAllocationMeasurement, ...] = Field(min_length=1)

    @model_validator(mode="after")
    def validate_traced_protocol(self) -> Self:
        """Order repetitions, and require a reproduced reviewed peak."""
        indices = [
            measurement.repetition_index for measurement in self.measurements
        ]
        if len(set(indices)) != len(indices):
            raise ValueError("repetition indices must be unique")
        if indices != sorted(indices):
            raise ValueError(
                "measurements must be ordered by repetition index"
            )
        if self.status is EvidenceStatus.REVIEWED:
            peaks = [
                measurement.peak_traced_bytes
                for measurement in self.measurements
            ]
            if len(self.measurements) < _MINIMUM_REVIEWED_TRACED_REPETITIONS:
                raise ValueError(
                    "reviewed traced evidence requires two repetitions"
                )
            if max(peaks) - min(peaks) > TRACED_PEAK_TOLERANCE_BYTES:
                raise ValueError(
                    "reviewed traced evidence requires one reproduced peak, "
                    "within a tenth of a mebibyte"
                )
        return self

validate_traced_protocol()

Order repetitions, and require a reproduced reviewed peak.

Source code in src/hebog/validation/evidence.py 
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
@model_validator(mode="after")
def validate_traced_protocol(self) -> Self:
    """Order repetitions, and require a reproduced reviewed peak."""
    indices = [
        measurement.repetition_index for measurement in self.measurements
    ]
    if len(set(indices)) != len(indices):
        raise ValueError("repetition indices must be unique")
    if indices != sorted(indices):
        raise ValueError(
            "measurements must be ordered by repetition index"
        )
    if self.status is EvidenceStatus.REVIEWED:
        peaks = [
            measurement.peak_traced_bytes
            for measurement in self.measurements
        ]
        if len(self.measurements) < _MINIMUM_REVIEWED_TRACED_REPETITIONS:
            raise ValueError(
                "reviewed traced evidence requires two repetitions"
            )
        if max(peaks) - min(peaks) > TRACED_PEAK_TOLERANCE_BYTES:
            raise ValueError(
                "reviewed traced evidence requires one reproduced peak, "
                "within a tenth of a mebibyte"
            )
    return self

TracedAllocationMeasurement

Bases: _EvidenceModel

One traced repetition of one input through the public finder.

peak_traced_bytes is the gate figure: the highest total of interpreter and NumPy allocations traced in the worker process, from before Hebog is imported until the products are written. finder_peak_traced_bytes is the peak over the find_sources call alone, measured from the same tracing session after its peak was reset, and import_traced_bytes is what the imports still hold when that call begins. Both make a peak comparable with one measured over a different span. peak_rss_bytes is an envelope, not a threshold: it varies with machine load, and tracing inflates it further.

Source code in src/hebog/validation/evidence.py 
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
class TracedAllocationMeasurement(_EvidenceModel):
    """One traced repetition of one input through the public finder.

    ``peak_traced_bytes`` is the gate figure: the highest total of
    interpreter and NumPy allocations traced in the worker process, from
    before Hebog is imported until the products are written.
    ``finder_peak_traced_bytes`` is the peak over the ``find_sources`` call
    alone, measured from the same tracing session after its peak was reset,
    and ``import_traced_bytes`` is what the imports still hold when that call
    begins. Both make a peak comparable with one measured over a different
    span. ``peak_rss_bytes`` is an envelope, not a threshold: it varies with
    machine load, and tracing inflates it further.
    """

    repetition_index: int = Field(ge=0)
    peak_traced_bytes: int = Field(gt=0)
    finder_peak_traced_bytes: int = Field(gt=0)
    import_traced_bytes: int = Field(gt=0)
    peak_rss_bytes: int = Field(ge=0)
    traced_wall_seconds: float = Field(ge=0, allow_inf_nan=False)
    source_count: int = Field(ge=0)
    gaussian_component_count: int = Field(ge=0)

    @model_validator(mode="after")
    def validate_peaks(self) -> Self:
        """Require spans that one tracing session can have produced."""
        if self.finder_peak_traced_bytes > self.peak_traced_bytes:
            raise ValueError("the finder peak cannot exceed the process peak")
        if self.import_traced_bytes > self.finder_peak_traced_bytes:
            raise ValueError("the import floor cannot exceed the finder peak")
        return self

validate_peaks()

Require spans that one tracing session can have produced.

Source code in src/hebog/validation/evidence.py 
419
420
421
422
423
424
425
426
@model_validator(mode="after")
def validate_peaks(self) -> Self:
    """Require spans that one tracing session can have produced."""
    if self.finder_peak_traced_bytes > self.peak_traced_bytes:
        raise ValueError("the finder peak cannot exceed the process peak")
    if self.import_traced_bytes > self.finder_peak_traced_bytes:
        raise ValueError("the import floor cannot exceed the finder peak")
    return self

UnavailableMetric

Bases: _EvidenceModel

One unavailable optional instrument and its explicit reason.

Source code in src/hebog/validation/evidence.py 
203
204
205
206
207
208
209
210
211
212
213
214
class UnavailableMetric(_EvidenceModel):
    """One unavailable optional instrument and its explicit reason."""

    metric: OptionalMetricName
    reason: str = Field(min_length=1)

    @model_validator(mode="after")
    def validate_reason(self) -> Self:
        """Reject whitespace-only explanations."""
        if not self.reason.strip():
            raise ValueError("unavailable metric reason must not be blank")
        return self

validate_reason()

Reject whitespace-only explanations.

Source code in src/hebog/validation/evidence.py 
209
210
211
212
213
214
@model_validator(mode="after")
def validate_reason(self) -> Self:
    """Reject whitespace-only explanations."""
    if not self.reason.strip():
        raise ValueError("unavailable metric reason must not be blank")
    return self

WorkloadClass

Bases: str, Enum

Scientific work represented by a benchmark dataset.

Source code in src/hebog/validation/evidence.py 
58
59
60
61
62
63
class WorkloadClass(str, Enum):
    """Scientific work represented by a benchmark dataset."""

    EMPTY_SPARSE = "empty-or-sparse"
    NORMAL = "normal"
    DENSE_EXTENDED = "dense-or-extended"

canonical_evidence_json(document)

Serialize evidence deterministically with a final newline.

Source code in src/hebog/validation/evidence.py 
499
500
501
502
503
504
505
506
507
508
509
510
def canonical_evidence_json(document: EvidenceDocument) -> str:
    """Serialize evidence deterministically with a final newline."""
    payload = document.model_dump(mode="json")
    return (
        json.dumps(
            payload,
            allow_nan=False,
            indent=2,
            sort_keys=True,
        )
        + "\n"
    )

load_evidence(path)

Load and validate one benchmark or scientific evidence document.

Source code in src/hebog/validation/evidence.py 
523
524
525
526
def load_evidence(path: Path) -> EvidenceDocument:
    """Load and validate one benchmark or scientific evidence document."""
    payload = path.read_text(encoding="utf-8")
    return _EVIDENCE_ADAPTER.validate_json(payload)

write_evidence(path, document)

Atomically write one validated evidence document.

Source code in src/hebog/validation/evidence.py 
513
514
515
516
517
518
519
520
def write_evidence(path: Path, document: EvidenceDocument) -> None:
    """Atomically write one validated evidence document."""
    temporary_path = path.with_suffix(f"{path.suffix}.tmp")
    temporary_path.write_text(
        canonical_evidence_json(document),
        encoding="utf-8",
    )
    temporary_path.replace(path)

hebog.validation.overhead

Versioned evidence for warm one-tile framework overhead measurements.

OverheadEnvironment

Bases: _OverheadModel

Exact local environment used for an exploratory overhead probe.

Source code in src/hebog/validation/overhead.py 
64
65
66
67
68
69
70
71
72
class OverheadEnvironment(_OverheadModel):
    """Exact local environment used for an exploratory overhead probe."""

    python: str = Field(min_length=1)
    platform: str = Field(min_length=1)
    machine: str = Field(min_length=1)
    cpu_count: int | None = Field(ge=1)
    node_memory_bytes: int = Field(ge=1)
    dependency_versions: dict[str, str]

OverheadEvidence

Bases: _OverheadModel

Complete warm one-tile overhead observation.

Source code in src/hebog/validation/overhead.py 
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
class OverheadEvidence(_OverheadModel):
    """Complete warm one-tile overhead observation."""

    schema_version: Literal[1]
    status: Literal["exploratory"]
    captured_at: datetime
    source_commit: str = Field(pattern=_COMMIT_PATTERN)
    dataset_identifier: str = Field(pattern=r"^[a-z0-9]+(?:-[a-z0-9]+)*$")
    dataset_content_sha256: str = Field(pattern=_SHA256_PATTERN)
    shape_yx: tuple[int, int]
    performance_contract_sha256: str = Field(pattern=_SHA256_PATTERN)
    environment_sha256: str = Field(pattern=_SHA256_PATTERN)
    environment: OverheadEnvironment
    measurements: tuple[OverheadStatistics, ...]

    @model_validator(mode="after")
    def validate_document(self) -> Self:
        """Require timezone, geometry, and all six unique operations."""
        if (
            self.captured_at.tzinfo is None
            or self.captured_at.utcoffset() is None
        ):
            raise ValueError("captured_at must include a timezone")
        if any(dimension <= 0 for dimension in self.shape_yx):
            raise ValueError("shape_yx dimensions must be positive")
        operations = [item.operation for item in self.measurements]
        if len(set(operations)) != len(operations):
            raise ValueError("overhead operations must be unique")
        if set(operations) != {
            "configuration",
            "fits-io",
            "partition-planning",
            "serial-dispatch",
            "local-dispatch",
            "dask-dispatch",
        }:
            raise ValueError("overhead evidence requires all six operations")
        return self

validate_document()

Require timezone, geometry, and all six unique operations.

Source code in src/hebog/validation/overhead.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
@model_validator(mode="after")
def validate_document(self) -> Self:
    """Require timezone, geometry, and all six unique operations."""
    if (
        self.captured_at.tzinfo is None
        or self.captured_at.utcoffset() is None
    ):
        raise ValueError("captured_at must include a timezone")
    if any(dimension <= 0 for dimension in self.shape_yx):
        raise ValueError("shape_yx dimensions must be positive")
    operations = [item.operation for item in self.measurements]
    if len(set(operations)) != len(operations):
        raise ValueError("overhead operations must be unique")
    if set(operations) != {
        "configuration",
        "fits-io",
        "partition-planning",
        "serial-dispatch",
        "local-dispatch",
        "dask-dispatch",
    }:
        raise ValueError("overhead evidence requires all six operations")
    return self

OverheadStatistics

Bases: _OverheadModel

Repeated warm timings for one isolated framework operation.

Source code in src/hebog/validation/overhead.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
class OverheadStatistics(_OverheadModel):
    """Repeated warm timings for one isolated framework operation."""

    operation: OverheadOperation
    method: str = Field(min_length=1)
    warmup_repetitions: int = Field(ge=1)
    measured_repetitions: int = Field(ge=5)
    minimum_seconds: float = Field(ge=0, allow_inf_nan=False)
    median_seconds: float = Field(ge=0, allow_inf_nan=False)
    percentile_95_seconds: float = Field(ge=0, allow_inf_nan=False)
    maximum_seconds: float = Field(ge=0, allow_inf_nan=False)
    budget_seconds: float = Field(gt=0, allow_inf_nan=False)
    within_budget: bool

    @model_validator(mode="after")
    def validate_summary(self) -> Self:
        """Require ordered statistics and a truthful budget result."""
        if not (
            self.minimum_seconds
            <= self.median_seconds
            <= self.percentile_95_seconds
            <= self.maximum_seconds
        ):
            raise ValueError("overhead statistics are not ordered")
        if self.within_budget != (
            self.percentile_95_seconds <= self.budget_seconds
        ):
            raise ValueError(
                "within_budget does not match the 95th percentile"
            )
        return self

validate_summary()

Require ordered statistics and a truthful budget result.

Source code in src/hebog/validation/overhead.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
@model_validator(mode="after")
def validate_summary(self) -> Self:
    """Require ordered statistics and a truthful budget result."""
    if not (
        self.minimum_seconds
        <= self.median_seconds
        <= self.percentile_95_seconds
        <= self.maximum_seconds
    ):
        raise ValueError("overhead statistics are not ordered")
    if self.within_budget != (
        self.percentile_95_seconds <= self.budget_seconds
    ):
        raise ValueError(
            "within_budget does not match the 95th percentile"
        )
    return self

load_overhead_evidence(path)

Load and validate one overhead-evidence document.

Source code in src/hebog/validation/overhead.py 
131
132
133
134
135
def load_overhead_evidence(path: Path) -> OverheadEvidence:
    """Load and validate one overhead-evidence document."""
    return OverheadEvidence.model_validate_json(
        path.read_text(encoding="utf-8")
    )

write_overhead_evidence(path, evidence)

Atomically write deterministic overhead evidence.

Source code in src/hebog/validation/overhead.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
def write_overhead_evidence(path: Path, evidence: OverheadEvidence) -> None:
    """Atomically write deterministic overhead evidence."""
    payload = (
        json.dumps(
            evidence.model_dump(mode="json"),
            allow_nan=False,
            indent=2,
            sort_keys=True,
        )
        + "\n"
    )
    temporary_path = path.with_suffix(f"{path.suffix}.tmp")
    temporary_path.write_text(payload, encoding="utf-8")
    temporary_path.replace(path)