Performance and scalability contracts

The machine-readable gates are config/benchmarks/phase-0-performance.json and config/benchmarks/phase-0-scalability.json. They stop implementation and tuning from selecting favourable inputs or hardware. Both have frozen-provisional status: changing a gate needs a reviewed plan decision and a commit recording its rationale and evidence references, and passing the file schema does not claim the gate has been demonstrated. Matched PyBDSF reference timings are under config/baselines/, and the current position against every gate is on progress against goals.

Complete performance curve

The ladder is 256, 512, 1,024, 2,048, 3,000, 4,096, 8,000, 8,192, 10,000, 16,384, 30,000, 32,768, 65,536 and 100,000 pixels per side, each with empty-or-sparse, normal and dense-or-extended workloads. The near-duplicates keep operational anchors beside powers of two so partition and storage effects stay visible. When the fastest valid execution plan changes, the matrix gains the nearest reproducible case below and above the observed crossover.

Every affected tier compares a candidate with the previous reviewed Hebog baseline over one warm-up and at least five measured repetitions. A change is a regression when the lower one-sided 95% bootstrap bound of the new/previous median ratio exceeds 1.05 without an approved trade-off.

The deployment gate is the pybdsf_master record, against pinned PyBDSF master (c70103b) over the record's measured_phase, the source-finding part of Rapthor's filter_skymodel step from the task's start to LSMTool's filter_sources call, which the finder controls. The upper one-sided 95% bound of the Hebog/PyBDSF median wall-time ratio must be at most 0.50 on a dense sector, provisionally one where master finds at least 10,000 sources, and at most 1.05 on the rest (decision of 18 October 2026). The record holds both limits and the threshold, and the matched benchmark classes each sector by the sources in master's catalogue, which every measured run must agree on. The complete step, whose LSMTool filtering is the same for either finder, is reported beside it and never gated (decision of 16 October 2026). Released 1.14.1 is checked once before 1.0.0 and is not a gate in this file. The matched benchmark applies this rule to both finders in one container. The quick benchmark reads its repetition counts, the regression rule and the confidence level from this file; its master ratio is diagnostic, with no outcome, because the environments are not matched and its cases are not classed.

The warm one-tile framework budgets are 250 ms for configuration, 500 ms for FITS I/O, 10 ms for partition planning, 5 ms for serial dispatch, 50 ms for local dispatch and 500 ms for existing-client Dask dispatch. They isolate framework cost from scientific stage time.

Representative component budgets

These 3,000-square allocations guide profiling. They are neither a measured runtime nor a substitute for matched complete filter_skymodel evidence.

Component	Budget
FITS input, validation, beam, WCS	1.5 s
True-sky background and RMS	4.0 s
Detection, deblending, durable image products	3.5 s
Compact measurement and fitting	2.0 s
Multiscale processing and merge	6.0 s
Catalogue and filter outputs	2.0 s
Flat-noise branch, when admitted concurrently	4.0 s
Dask scheduling/transfer on critical path	2.0 s

The planned true-sky critical path is about 19 seconds, with flat-noise work hidden only when the resource envelope admits concurrency.

Extreme-image resource envelope

The 100,000-square case has two logical intensity inputs and three required image outputs: true-sky RMS, flat-noise RMS and the source-filter mask. Catalogue and filtered sky-model products are records, not planes. The planner may admit at most eight live float32 plane-equivalents across bounded buffers, and no worker may materialise a full extreme plane.

The provisional production profile is a 512 GiB node with four workers of eight threads: 64 GiB reserved for the operating system, scheduler and services, 128 GiB for concurrent pipeline work, and four 80 GiB worker limits, with normal worker peak memory at most 75% of that limit. Dask target/spill/pause/terminate fractions are 0.70/0.80/0.90/0.95, spill uses worker-local NVMe, and normal-run spill stays below 5% of logical input bytes. Qualification records the actual facility topology and never scales worker limits past admitted RAM to make a run pass.

Candidate square tile cores are 2,048, 4,096 and 8,192 pixels. A stage halo stays below a quarter of its core, or the stage selects a larger core or records a contract change. Batch sizing chooses the largest stage-valid batch below the pessimistic memory limit, then reduces batch size before the tile core. Resource choices never alter core ownership or scientific results.

Controlled node gates

Strong-scaling qualification preserves every result at 1, 10, 50, 100 and 200 nodes, with provisional maximum complete runtimes of 3,600, 600, 180, 120 and 90 seconds, scheduler overhead at most 10% of the critical path and at most 50,000 tasks in the graph. Occupancy and strong/weak efficiency floors are recorded per topology in the machine-readable contract. The plan's task 27 amends this to the development-machine tier and the 1.0.0 cluster benchmark of one Rapthor sector at 1, 2, 5 and 10 nodes, keeping the larger node counts as design targets. Only controlled multi-node evidence in the versioned evidence schema can demonstrate these gates.

hebog.validation.contracts

Versioned performance, scalability, behaviour, and scientific contracts.

These records describe gates and test obligations. They do not select an execution plan or import a scheduler, read science data, or run benchmarks.

CrossoverProbe

Bases: _ContractModel

Initial range in which an execution crossover must be measured.

Source code in src/hebog/validation/contracts.py

class CrossoverProbe(_ContractModel):
    """Initial range in which an execution crossover must be measured."""

    boundary: ExecutionBoundary
    lower_size_pixels: int = Field(ge=1)
    upper_size_pixels: int = Field(ge=1)

    @model_validator(mode="after")
    def validate_order(self) -> Self:
        """Require a non-empty ordered probe interval."""
        if self.lower_size_pixels >= self.upper_size_pixels:
            raise ValueError("crossover probe bounds must be increasing")
        return self

validate_order()

Require a non-empty ordered probe interval.

Source code in src/hebog/validation/contracts.py

@model_validator(mode="after")
def validate_order(self) -> Self:
    """Require a non-empty ordered probe interval."""
    if self.lower_size_pixels >= self.upper_size_pixels:
        raise ValueError("crossover probe bounds must be increasing")
    return self

ExecutionBoundary

Bases: str, Enum

A performance crossover that requires explicit bracketing cases.

Source code in src/hebog/validation/contracts.py

class ExecutionBoundary(str, Enum):
    """A performance crossover that requires explicit bracketing cases."""

    DIRECT_TO_LOCAL = "direct-to-local"
    LOCAL_TO_DASK = "local-to-dask"
    DIRECT_TO_CHUNKED_STORAGE = "direct-to-chunked-storage"

IslandScientificGate

Bases: _ContractModel

Minimum object recovery and overlap metrics for one Phase 3 lane.

Source code in src/hebog/validation/contracts.py

class IslandScientificGate(_ContractModel):
    """Minimum object recovery and overlap metrics for one Phase 3 lane."""

    minimum_completeness: float = Field(ge=0, le=1)
    minimum_reliability: float = Field(ge=0, le=1)
    minimum_median_intersection_over_union: float = Field(ge=0, le=1)
    minimum_matched_intersection_over_union: float = Field(ge=0, le=1)
    maximum_split_count: int = Field(ge=0)
    maximum_merge_count: int = Field(ge=0)

MaskScientificGate

Bases: _ContractModel

Minimum foreground-sensitive pixel metrics for one Phase 3 lane.

Source code in src/hebog/validation/contracts.py

class MaskScientificGate(_ContractModel):
    """Minimum foreground-sensitive pixel metrics for one Phase 3 lane."""

    minimum_precision: float = Field(ge=0, le=1)
    minimum_recall: float = Field(ge=0, le=1)
    minimum_intersection_over_union: float = Field(ge=0, le=1)

NodeScalingGate

Bases: _ContractModel

Runtime and efficiency limits at one controlled topology.

Source code in src/hebog/validation/contracts.py

class NodeScalingGate(_ContractModel):
    """Runtime and efficiency limits at one controlled topology."""

    worker_nodes: int = Field(ge=1)
    maximum_runtime_seconds: float = Field(gt=0)
    maximum_scheduler_overhead_fraction: float = Field(ge=0, lt=1)
    minimum_worker_occupancy_fraction: float = Field(ge=0, le=1)
    minimum_strong_scaling_efficiency: float | None = Field(default=None, gt=0)
    minimum_weak_scaling_efficiency: float | None = Field(default=None, gt=0)

OneTileOverheadBudget

Bases: _ContractModel

Maximum framework overhead for a warm one-tile request.

Source code in src/hebog/validation/contracts.py

class OneTileOverheadBudget(_ContractModel):
    """Maximum framework overhead for a warm one-tile request."""

    configuration_seconds: float = Field(ge=0)
    fits_io_seconds: float = Field(ge=0)
    partition_planning_seconds: float = Field(ge=0)
    serial_dispatch_seconds: float = Field(ge=0)
    local_dispatch_seconds: float = Field(ge=0)
    dask_dispatch_seconds: float = Field(ge=0)

PerformanceMatrixContract

Bases: _ContractModel

Frozen logarithmic size and scientific-work performance matrix.

Source code in src/hebog/validation/contracts.py

class PerformanceMatrixContract(_ContractModel):
    """Frozen logarithmic size and scientific-work performance matrix."""

    schema_version: Literal[3]
    matrix_id: str = Field(pattern=r"^[a-z0-9]+(?:-[a-z0-9]+)*$")
    status: Literal["frozen-provisional"]
    sizes_pixels: tuple[int, ...] = Field(min_length=2)
    workload_classes: tuple[WorkloadClass, ...]
    initial_crossover_probes: tuple[CrossoverProbe, ...]
    crossover_bracketing_rule: str = Field(min_length=1)
    pybdsf_master: PyBdsfMasterGate
    previous_hebog: PreviousHebogGate
    one_tile_overhead_budget: OneTileOverheadBudget

    @model_validator(mode="after")
    def validate_matrix(self) -> Self:
        """Protect matrix coverage and comparison semantics."""
        if tuple(sorted(set(self.sizes_pixels))) != self.sizes_pixels:
            raise ValueError("performance sizes must be unique and increasing")
        if (
            self.sizes_pixels[0] != _MINIMUM_MATRIX_SIZE
            or self.sizes_pixels[-1] != _MAXIMUM_MATRIX_SIZE
        ):
            raise ValueError("performance matrix must span 256 to 100000")
        if set(self.workload_classes) != set(WorkloadClass):
            raise ValueError(
                "performance matrix must include every workload class"
            )
        if len(set(self.workload_classes)) != len(self.workload_classes):
            raise ValueError("performance workload classes must be unique")
        boundaries = [
            probe.boundary for probe in self.initial_crossover_probes
        ]
        if set(boundaries) != set(ExecutionBoundary):
            raise ValueError(
                "every execution boundary requires an initial probe"
            )
        if len(set(boundaries)) != len(boundaries):
            raise ValueError("execution boundary probes must be unique")
        return self

validate_matrix()

Protect matrix coverage and comparison semantics.

Source code in src/hebog/validation/contracts.py

@model_validator(mode="after")
def validate_matrix(self) -> Self:
    """Protect matrix coverage and comparison semantics."""
    if tuple(sorted(set(self.sizes_pixels))) != self.sizes_pixels:
        raise ValueError("performance sizes must be unique and increasing")
    if (
        self.sizes_pixels[0] != _MINIMUM_MATRIX_SIZE
        or self.sizes_pixels[-1] != _MAXIMUM_MATRIX_SIZE
    ):
        raise ValueError("performance matrix must span 256 to 100000")
    if set(self.workload_classes) != set(WorkloadClass):
        raise ValueError(
            "performance matrix must include every workload class"
        )
    if len(set(self.workload_classes)) != len(self.workload_classes):
        raise ValueError("performance workload classes must be unique")
    boundaries = [
        probe.boundary for probe in self.initial_crossover_probes
    ]
    if set(boundaries) != set(ExecutionBoundary):
        raise ValueError(
            "every execution boundary requires an initial probe"
        )
    if len(set(boundaries)) != len(boundaries):
        raise ValueError("execution boundary probes must be unique")
    return self

PhaseFourAssociationContract

Bases: _ContractModel

Provisional compact component and source association policy.

Source code in src/hebog/validation/contracts.py

class PhaseFourAssociationContract(_ContractModel):
    """Provisional compact component and source association policy."""

    region_membership: Literal["worker-local-watershed-labels"]
    compact_component_policy: Literal[
        "one-fitted-gaussian-per-deblended-region"
    ]
    compact_source_policy: Literal["one-source-per-deblended-region"]
    parent_policy: Literal["retain-reconciled-parent-island"]
    unsupported_region_policy: Literal["explicit-later-phase-deferral"]
    identifier_policy: Literal[
        "canonical-global-topology-and-association-order"
    ]
    truth_resolvability_policy: Literal["distinct-eligible-observed-maximum"]
    unresolved_truth_policy: Literal["explicit-group-centroid-and-total-flux"]
    joint_model_policy: Literal[
        "deferred-until-identifiability-and-reliability-evidence"
    ]

PhaseFourCatalogueGate

Bases: _ContractModel

Catalogue non-inferiority margins for one governed dataset lane.

Source code in src/hebog/validation/contracts.py

class PhaseFourCatalogueGate(_ContractModel):
    """Catalogue non-inferiority margins for one governed dataset lane."""

    detection_population: Literal["declared-compact-snr-at-least-10"]
    position_and_flux_population: Literal["isolated-compact-snr-at-least-10"]
    shape_population: Literal["eligible-compact-snr-at-least-10"]
    association_population: Literal[
        "declared-compact-associations-snr-at-least-10"
    ]
    outlier_population: Literal["matched-compact-snr-at-least-10"]
    absolute_median_policy: Literal["gate", "report-only"] = "gate"
    absolute_tail_policy: Literal["gate", "report-only"]
    minimum_completeness: float = Field(ge=0, le=1)
    minimum_reliability: float = Field(ge=0, le=1)
    maximum_median_position_beams: float = Field(ge=0)
    maximum_percentile_95_position_beams: float = Field(ge=0)
    maximum_median_peak_flux_fractional_difference: float = Field(ge=0)
    maximum_percentile_95_peak_flux_fractional_difference: float = Field(ge=0)
    maximum_median_integrated_flux_fractional_difference: float = Field(ge=0)
    maximum_percentile_95_integrated_flux_fractional_difference: float = Field(
        ge=0
    )
    maximum_median_fitted_axis_fractional_difference: float = Field(ge=0)
    maximum_percentile_95_fitted_axis_fractional_difference: float = Field(
        ge=0
    )
    maximum_median_deconvolved_axis_fractional_difference: float = Field(ge=0)
    maximum_percentile_95_deconvolved_axis_fractional_difference: float = (
        Field(ge=0)
    )
    maximum_median_position_angle_difference_degrees: float = Field(ge=0)
    maximum_percentile_95_position_angle_difference_degrees: float = Field(
        ge=0
    )
    minimum_point_source_specificity: float = Field(ge=0, le=1)
    minimum_clear_resolved_classification_recall: float = Field(ge=0, le=1)
    minimum_association_pair_precision: float = Field(ge=0, le=1)
    minimum_association_pair_recall: float = Field(ge=0, le=1)
    minimum_fitted_shape_availability: float = Field(ge=0, le=1)
    minimum_deconvolution_classification_availability: float = Field(
        ge=0, le=1
    )
    minimum_resolved_deconvolved_shape_availability: float = Field(ge=0, le=1)
    minimum_association_identity_availability: float = Field(ge=0, le=1)
    minimum_position_flux_uncertainty_availability: float = Field(ge=0, le=1)
    maximum_catastrophic_outlier_fraction: float = Field(ge=0, le=1)

    @model_validator(mode="after")
    def validate_tail_margins(self) -> Self:
        """Keep every 95th-percentile ceiling no tighter than its median."""
        margin_pairs = (
            (
                self.maximum_median_position_beams,
                self.maximum_percentile_95_position_beams,
            ),
            (
                self.maximum_median_peak_flux_fractional_difference,
                self.maximum_percentile_95_peak_flux_fractional_difference,
            ),
            (
                self.maximum_median_integrated_flux_fractional_difference,
                self.maximum_percentile_95_integrated_flux_fractional_difference,
            ),
            (
                self.maximum_median_fitted_axis_fractional_difference,
                self.maximum_percentile_95_fitted_axis_fractional_difference,
            ),
            (
                self.maximum_median_deconvolved_axis_fractional_difference,
                self.maximum_percentile_95_deconvolved_axis_fractional_difference,
            ),
            (
                self.maximum_median_position_angle_difference_degrees,
                self.maximum_percentile_95_position_angle_difference_degrees,
            ),
        )
        if any(tail < median for median, tail in margin_pairs):
            raise ValueError(
                "95th-percentile catalogue margin cannot be tighter than "
                "median"
            )
        return self

validate_tail_margins()

Keep every 95th-percentile ceiling no tighter than its median.

Source code in src/hebog/validation/contracts.py

@model_validator(mode="after")
def validate_tail_margins(self) -> Self:
    """Keep every 95th-percentile ceiling no tighter than its median."""
    margin_pairs = (
        (
            self.maximum_median_position_beams,
            self.maximum_percentile_95_position_beams,
        ),
        (
            self.maximum_median_peak_flux_fractional_difference,
            self.maximum_percentile_95_peak_flux_fractional_difference,
        ),
        (
            self.maximum_median_integrated_flux_fractional_difference,
            self.maximum_percentile_95_integrated_flux_fractional_difference,
        ),
        (
            self.maximum_median_fitted_axis_fractional_difference,
            self.maximum_percentile_95_fitted_axis_fractional_difference,
        ),
        (
            self.maximum_median_deconvolved_axis_fractional_difference,
            self.maximum_percentile_95_deconvolved_axis_fractional_difference,
        ),
        (
            self.maximum_median_position_angle_difference_degrees,
            self.maximum_percentile_95_position_angle_difference_degrees,
        ),
    )
    if any(tail < median for median, tail in margin_pairs):
        raise ValueError(
            "95th-percentile catalogue margin cannot be tighter than "
            "median"
        )
    return self

PhaseFourCoordinateContract

Bases: _ContractModel

Pixel, celestial, shape, and uncertainty coordinate conventions.

Source code in src/hebog/validation/contracts.py

class PhaseFourCoordinateContract(_ContractModel):
    """Pixel, celestial, shape, and uncertainty coordinate conventions."""

    celestial_frame: Literal["icrs"]
    pixel_origin: Literal["zero-based-pixel-centres"]
    pixel_coordinate_order: Literal["x-y"]
    transform: Literal["astropy-local-tangent-plane-jacobian"]
    position_angle: Literal["east-of-north-modulo-180"]
    uncertainty_interpretation: Literal["one-standard-deviation"]

PhaseFourEligibilityContract

Bases: _ContractModel

Reference-selected populations and missing-value gate semantics.

Source code in src/hebog/validation/contracts.py

class PhaseFourEligibilityContract(_ContractModel):
    """Reference-selected populations and missing-value gate semantics."""

    population_selection: Literal["reference-or-injected-truth-only"]
    missing_candidate_value: Literal["counts-as-unavailable-not-excluded"]
    fitted_shape_eligibility: Literal["reference-fitted-shape-available"]
    deconvolved_shape_eligibility: Literal["reference-resolved-with-shape"]
    position_angle_eligibility: Literal[
        "eligible-reference-shape-axis-ratio-at-least-minimum"
    ]
    position_angle_minimum_axis_ratio: float = Field(gt=1)
    association_eligibility: Literal["reference-parent-association-declared"]
    uncertainty_eligibility: Literal["reference-compact-snr-at-least-10"]
    availability_reporting: Literal["required-for-every-gated-field"]

PhaseFourExtensionClassification

Bases: _ContractModel

Frozen uncertainty-aware point/extended classification policy.

Source code in src/hebog/validation/contracts.py

class PhaseFourExtensionClassification(_ContractModel):
    """Frozen uncertainty-aware point/extended classification policy."""

    method: Literal["integrated-to-peak-ratio-uncertainty"]
    significance_sigma: float = Field(ge=2.0, le=5.0)
    clear_resolved_minimum_area_ratio: float = Field(ge=2.0)
    clear_resolved_minimum_signal_to_noise: float = Field(ge=10.0)
    point_source_population: Literal["shape-unresolved"]
    clear_resolved_population: Literal["shape-clear-resolved"]
    marginal_resolved_population: Literal[
        "shape-marginal-resolved-report-only"
    ]
    marginal_resolved_integrated_flux_catastrophic_rate: Literal["report-only"]
    unresolved_integrated_flux: Literal["peak-flux-density"]
    resolved_integrated_flux_uncertainty: Literal["report-only"]
    missing_uncertainty: Literal[
        "geometric-noiseless-reference-otherwise-unavailable"
    ]

PhaseFourFailureContract

Bases: _ContractModel

Scientific absence and compatibility sentinel semantics.

Source code in src/hebog/validation/contracts.py

class PhaseFourFailureContract(_ContractModel):
    """Scientific absence and compatibility sentinel semantics."""

    unresolved_deconvolution: Literal[
        "null-shape-with-unresolved-quality-flag"
    ]
    compatibility_unresolved_shape: Literal["adapter-zero-axes-only"]
    unavailable_uncertainty: Literal["null-with-quality-flag"]
    invalid_measurement: Literal["typed-unavailable-result"]
    omitted_deferred_region: Literal["forbidden-in-complete-result"]

PhaseFourFittingContract

Bases: _ContractModel

Evidence order for nonlinear and selective fitting.

Source code in src/hebog/validation/contracts.py

class PhaseFourFittingContract(_ContractModel):
    """Evidence order for nonlinear and selective fitting."""

    reference_policy: Literal["fit-all-admitted-compact-regions"]
    selective_policy: Literal["fit-all-reference-before-selection"]
    moment_policy: Literal["serial-oracle-and-fit-initializer"]
    implementation_selection: Literal[
        "established-library-science-and-complete-stage-evidence"
    ]

PhaseFourMeasurementContract

Bases: _ContractModel

Versioned Phase 4 compact measurement contract.

Source code in src/hebog/validation/contracts.py

class PhaseFourMeasurementContract(_ContractModel):
    """Versioned Phase 4 compact measurement contract."""

    schema_version: Literal[2]
    contract_id: Literal["phase-4-measurement"]
    status: Literal["frozen-provisional", "reviewed-provisional"]
    scope: PhaseFourScopeContract
    measurements: PhaseFourMeasurementSemantics
    coordinates: PhaseFourCoordinateContract
    association: PhaseFourAssociationContract
    failures: PhaseFourFailureContract
    fitting: PhaseFourFittingContract
    eligibility: PhaseFourEligibilityContract
    required_analytic_failure_cases: tuple[
        Literal[
            "fit-non-convergence",
            "marginal-deconvolution",
            "non-finite-owned-pixels",
            "non-positive-measurement",
            "singular-covariance",
            "underdetermined-region",
        ],
        ...,
    ] = Field(min_length=6)
    scientific_basis: tuple[str, ...] = Field(min_length=4)
    qualification_policy: Literal["freeze-before-result-inspection"]
    human_scientific_review: Literal["required-before-stable-default"]

    @model_validator(mode="after")
    def validate_scientific_basis(self) -> Self:
        """Require distinct immutable primary-source links."""
        if (
            len(self.required_analytic_failure_cases)
            != _PHASE_FOUR_ANALYTIC_FAILURE_CASE_COUNT
            or len(set(self.required_analytic_failure_cases))
            != _PHASE_FOUR_ANALYTIC_FAILURE_CASE_COUNT
        ):
            raise ValueError(
                "every Phase 4 analytic failure case must appear exactly once"
            )
        if len(set(self.scientific_basis)) != len(self.scientific_basis):
            raise ValueError("scientific basis links must be unique")
        if not all(
            source.startswith("https://") for source in self.scientific_basis
        ):
            raise ValueError("scientific basis links must use HTTPS")
        return self

validate_scientific_basis()

Require distinct immutable primary-source links.

Source code in src/hebog/validation/contracts.py

@model_validator(mode="after")
def validate_scientific_basis(self) -> Self:
    """Require distinct immutable primary-source links."""
    if (
        len(self.required_analytic_failure_cases)
        != _PHASE_FOUR_ANALYTIC_FAILURE_CASE_COUNT
        or len(set(self.required_analytic_failure_cases))
        != _PHASE_FOUR_ANALYTIC_FAILURE_CASE_COUNT
    ):
        raise ValueError(
            "every Phase 4 analytic failure case must appear exactly once"
        )
    if len(set(self.scientific_basis)) != len(self.scientific_basis):
        raise ValueError("scientific basis links must be unique")
    if not all(
        source.startswith("https://") for source in self.scientific_basis
    ):
        raise ValueError("scientific basis links must use HTTPS")
    return self

PhaseFourMeasurementSemantics

Bases: _ContractModel

Frozen meanings for compact catalogue measurements.

Source code in src/hebog/validation/contracts.py

class PhaseFourMeasurementSemantics(_ContractModel):
    """Frozen meanings for compact catalogue measurements."""

    pixel_solid_angle: Literal[
        "absolute-local-tangent-plane-jacobian-determinant"
    ]
    restoring_beam_solid_angle: Literal[
        "pi-major-fwhm-times-minor-fwhm-over-four-ln-two"
    ]
    island_integrated_flux: Literal[
        "owned-valid-pixel-sum-times-pixel-area-over-beam-area"
    ]
    island_local_rms: Literal["mean-rms-over-owned-valid-pixels"]
    island_mean_brightness: Literal[
        "mean-background-subtracted-brightness-over-owned-valid-pixels"
    ]
    component_peak_flux: Literal["fitted-gaussian-amplitude"]
    component_integrated_flux: Literal[
        "peak-for-unresolved-otherwise-peak-times-fitted-gaussian-area-over-beam-area"
    ]
    component_local_rms: Literal["bilinear-rms-map-at-fitted-centroid"]
    source_flux: Literal["associated-component-flux-for-compact-scope"]

PhaseFourOutlierDefinition

Bases: _ContractModel

Observable thresholds for one catastrophic matched-row outlier.

Source code in src/hebog/validation/contracts.py

class PhaseFourOutlierDefinition(_ContractModel):
    """Observable thresholds for one catastrophic matched-row outlier."""

    position_beams: float = Field(gt=0)
    peak_flux_fractional_difference: float = Field(gt=0)
    integrated_flux_fractional_difference: float = Field(gt=0)
    fitted_axis_fractional_difference: float = Field(gt=0)
    deconvolved_axis_fractional_difference: float = Field(gt=0)

PhaseFourScientificGates

Bases: _ContractModel

Versioned Phase 4 catalogue and uncertainty margins.

Source code in src/hebog/validation/contracts.py

class PhaseFourScientificGates(_ContractModel):
    """Versioned Phase 4 catalogue and uncertainty margins."""

    schema_version: Literal[2]
    contract_id: Literal["phase-4-scientific-gates"]
    status: Literal["frozen-provisional", "reviewed-provisional"]
    confidence_level: float = Field(gt=0, lt=1)
    low_snr_threshold_crossings: Literal["report-only"]
    shape_uncertainty: Literal["report-only"]
    noisy_source_decision: Literal[
        "snr-stratified-confidence-intervals-and-catastrophic-rate"
    ]
    compact_reference: PhaseFourCatalogueGate
    generated_regression: PhaseFourCatalogueGate
    heldout_qualification: PhaseFourCatalogueGate
    extension_classification: PhaseFourExtensionClassification
    unresolved_group: PhaseFourUnresolvedGroupGate
    uncertainty: PhaseFourUncertaintyGate
    catastrophic_outlier: PhaseFourOutlierDefinition

PhaseFourScopeContract

Bases: _ContractModel

Supported scientific image and unit scope for compact measurement.

Source code in src/hebog/validation/contracts.py

class PhaseFourScopeContract(_ContractModel):
    """Supported scientific image and unit scope for compact measurement."""

    image_kind: Literal["mfs-stokes-i"]
    measurement_plane: Literal["background-subtracted-primary-beam-corrected"]
    brightness_unit: Literal["Jy/beam"]
    frequency_semantics: Literal["reference-frequency-only"]
    invalid_pixels: Literal["excluded-from-membership-and-measurement"]

PhaseFourUncertaintyGate

Bases: _ContractModel

Calibration gates for reported one-standard-deviation errors.

Source code in src/hebog/validation/contracts.py

class PhaseFourUncertaintyGate(_ContractModel):
    """Calibration gates for reported one-standard-deviation errors."""

    nominal_coverage: float = Field(gt=0, lt=1)
    maximum_absolute_coverage_difference: float = Field(gt=0, lt=1)
    maximum_absolute_mean_normalized_residual: float = Field(gt=0)
    minimum_normalized_residual_standard_deviation: float = Field(gt=0)
    maximum_normalized_residual_standard_deviation: float = Field(gt=0)
    minimum_samples_per_stratum: int = Field(ge=200)
    confidence_interval_level: float = Field(gt=0, lt=1)
    equivalence_rule: Literal["entire-confidence-interval-within-margins"]
    coverage_interval: Literal[
        "wilson-score",
        "cluster-robust-student-t",
    ]
    mean_interval: Literal[
        "student-t",
        "cluster-robust-student-t",
    ]
    dispersion_interval: Literal[
        "scipy-bca-bootstrap-fixed-seed",
        "cluster-percentile-bootstrap-fixed-seed",
    ]
    bootstrap_resamples: int = Field(ge=10_000)
    bootstrap_seed: int = Field(ge=0)
    insufficient_samples: Literal["report-only"]

    @model_validator(mode="after")
    def validate_standard_deviation_range(self) -> Self:
        """Require an increasing calibration-dispersion interval."""
        if (
            self.minimum_normalized_residual_standard_deviation
            >= self.maximum_normalized_residual_standard_deviation
        ):
            raise ValueError(
                "uncertainty standard-deviation bounds must be increasing"
            )
        return self

validate_standard_deviation_range()

Require an increasing calibration-dispersion interval.

Source code in src/hebog/validation/contracts.py

@model_validator(mode="after")
def validate_standard_deviation_range(self) -> Self:
    """Require an increasing calibration-dispersion interval."""
    if (
        self.minimum_normalized_residual_standard_deviation
        >= self.maximum_normalized_residual_standard_deviation
    ):
        raise ValueError(
            "uncertainty standard-deviation bounds must be increasing"
        )
    return self

PhaseFourUnresolvedGroupGate

Bases: _ContractModel

Frozen-provisional margins for one observable unresolved group.

Source code in src/hebog/validation/contracts.py

class PhaseFourUnresolvedGroupGate(_ContractModel):
    """Frozen-provisional margins for one observable unresolved group."""

    status: Literal["frozen-provisional", "reviewed-provisional"]
    population: Literal["declared-unresolved-association-groups"]
    minimum_completeness: float = Field(ge=0, le=1)
    maximum_median_position_beams: float = Field(ge=0)
    maximum_percentile_95_position_beams: float = Field(ge=0)
    maximum_median_integrated_flux_fractional_difference: float = Field(ge=0)
    maximum_percentile_95_integrated_flux_fractional_difference: float = Field(
        ge=0
    )

    @model_validator(mode="after")
    def validate_tail_margins(self) -> Self:
        """Require group tail ceilings to include their median ceilings."""
        if (
            self.maximum_percentile_95_position_beams
            < self.maximum_median_position_beams
            or self.maximum_percentile_95_integrated_flux_fractional_difference
            < self.maximum_median_integrated_flux_fractional_difference
        ):
            raise ValueError(
                "95th-percentile unresolved-group margin cannot be tighter "
                "than median"
            )
        return self

validate_tail_margins()

Require group tail ceilings to include their median ceilings.

Source code in src/hebog/validation/contracts.py

@model_validator(mode="after")
def validate_tail_margins(self) -> Self:
    """Require group tail ceilings to include their median ceilings."""
    if (
        self.maximum_percentile_95_position_beams
        < self.maximum_median_position_beams
        or self.maximum_percentile_95_integrated_flux_fractional_difference
        < self.maximum_median_integrated_flux_fractional_difference
    ):
        raise ValueError(
            "95th-percentile unresolved-group margin cannot be tighter "
            "than median"
        )
    return self

PhaseThreeLaneGate

Bases: _ContractModel

Mask and connected-object gates for one governed dataset lane.

Source code in src/hebog/validation/contracts.py

class PhaseThreeLaneGate(_ContractModel):
    """Mask and connected-object gates for one governed dataset lane."""

    mask: MaskScientificGate
    islands: IslandScientificGate

PhaseThreeScientificGates

Bases: _ContractModel

Reviewed foreground-sensitive Phase 3 non-inferiority margins.

Source code in src/hebog/validation/contracts.py

class PhaseThreeScientificGates(_ContractModel):
    """Reviewed foreground-sensitive Phase 3 non-inferiority margins."""

    schema_version: Literal[1]
    contract_id: Literal["phase-3-scientific-gates"]
    status: Literal["reviewed-provisional"]
    confidence_level: float = Field(gt=0, lt=1)
    low_snr_threshold_crossings: Literal["report-only"]
    compact_reference: PhaseThreeLaneGate
    generated_regression: PhaseThreeLaneGate
    heldout_qualification: PhaseThreeLaneGate

PreviousHebogGate

Bases: _ContractModel

Statistical rule for comparing consecutive Hebog baselines.

Source code in src/hebog/validation/contracts.py

class PreviousHebogGate(_ContractModel):
    """Statistical rule for comparing consecutive Hebog baselines."""

    schema_version: Literal[1]
    warmup_repetitions: int = Field(ge=1)
    minimum_measured_repetitions: int = Field(ge=5)
    confidence_level: float = Field(gt=0, lt=1)
    bootstrap_resamples: int = Field(ge=1_000)
    regression_ratio_lower_bound: float = Field(gt=1)

PublicBehaviour

Bases: _ContractModel

One frozen observable behaviour and the test that holds it.

test_node_id is the pytest node ID of one test function, and its directory is the test's lane. An implemented behaviour names the ordinary test that asserts it. An unimplemented one names a strict-xfail placeholder: when the behaviour starts to pass, the placeholder fails until it becomes a normal assertion and its status implemented.

Source code in src/hebog/validation/contracts.py

class PublicBehaviour(_ContractModel):
    """One frozen observable behaviour and the test that holds it.

    ``test_node_id`` is the pytest node ID of one test function, and its
    directory is the test's lane. An implemented behaviour names the ordinary
    test that asserts it. An unimplemented one names a strict-xfail
    placeholder: when the behaviour starts to pass, the placeholder fails
    until it becomes a normal assertion and its status ``implemented``.
    """

    identifier: str = Field(pattern=r"^[a-z0-9]+(?:-[a-z0-9]+)*$")
    scenario: str = Field(min_length=1)
    status: Literal["implemented", "strict-xfail"]
    test_node_id: str = Field(
        pattern=r"^tests/[a-z]+/test_[a-z0-9_]+\.py::test_[a-z0-9_]+$"
    )

PublicBehaviourManifest

Bases: _ContractModel

Frozen list of public behaviours that drive red-green-refactor work.

Source code in src/hebog/validation/contracts.py

class PublicBehaviourManifest(_ContractModel):
    """Frozen list of public behaviours that drive red-green-refactor work."""

    schema_version: Literal[2]
    manifest_id: str = Field(pattern=r"^[a-z0-9]+(?:-[a-z0-9]+)*$")
    status: Literal["frozen-provisional"]
    behaviours: tuple[PublicBehaviour, ...] = Field(min_length=1)

    @model_validator(mode="after")
    def validate_unique_behaviours(self) -> Self:
        """Keep scenario and test ownership unambiguous."""
        identifiers = [behaviour.identifier for behaviour in self.behaviours]
        node_ids = [behaviour.test_node_id for behaviour in self.behaviours]
        if len(set(identifiers)) != len(identifiers):
            raise ValueError("public behaviour identifiers must be unique")
        if len(set(node_ids)) != len(node_ids):
            raise ValueError("public behaviour test node IDs must be unique")
        return self

validate_unique_behaviours()

Keep scenario and test ownership unambiguous.

Source code in src/hebog/validation/contracts.py

@model_validator(mode="after")
def validate_unique_behaviours(self) -> Self:
    """Keep scenario and test ownership unambiguous."""
    identifiers = [behaviour.identifier for behaviour in self.behaviours]
    node_ids = [behaviour.test_node_id for behaviour in self.behaviours]
    if len(set(identifiers)) != len(identifiers):
        raise ValueError("public behaviour identifiers must be unique")
    if len(set(node_ids)) != len(node_ids):
        raise ValueError("public behaviour test node IDs must be unique")
    return self

PyBdsfMasterGate

Bases: _ContractModel

Deployment runtime gate against one pinned PyBDSF master commit.

A candidate passes a sector when the upper one-sided confidence bound of its median wall-time ratio to the pinned commit is at most its sector class's limit over the measured_phase: the source-finding part of Rapthor's filter_skymodel step, from the task's start to LSMTool's filter_sources call, which the finder controls. The complete step is reported beside it, never gated (decision of 16 October 2026). A sector is dense, where PyBDSF is slow, when master's catalogue holds at least dense_minimum_sources sources; Hebog must take at most dense_maximum_ratio of its time there and no longer than sparse_maximum_ratio of it on the rest (decision of 18 October 2026).

Source code in src/hebog/validation/contracts.py

class PyBdsfMasterGate(_ContractModel):
    """Deployment runtime gate against one pinned PyBDSF ``master`` commit.

    A candidate passes a sector when the upper one-sided confidence bound
    of its median wall-time ratio to the pinned commit is at most its
    sector class's limit over the ``measured_phase``: the source-finding
    part of Rapthor's ``filter_skymodel`` step, from the task's start to
    LSMTool's ``filter_sources`` call, which the finder controls. The
    complete step is reported beside it, never gated (decision of
    16 October 2026). A sector is dense, where PyBDSF is slow, when
    ``master``'s catalogue holds at least ``dense_minimum_sources``
    sources; Hebog must take at most ``dense_maximum_ratio`` of its time
    there and no longer than ``sparse_maximum_ratio`` of it on the rest
    (decision of 18 October 2026).
    """

    commit_sha: str = Field(pattern=r"^[0-9a-f]{40}$")
    dense_minimum_sources: int = Field(gt=0)
    dense_maximum_ratio: float = Field(gt=0, lt=1)
    sparse_maximum_ratio: float = Field(ge=1)
    confidence_level: float = Field(gt=0, lt=1)
    bound: Literal["upper-one-sided"]
    measured_phase: Literal["finder"]

    def sector_class(self, master_source_count: int) -> SectorClass:
        """Return the class of a sector where ``master`` found this many."""
        if master_source_count >= self.dense_minimum_sources:
            return "dense"
        return "sparse"

    def maximum_ratio(self, master_source_count: int) -> float:
        """Return the limit for a sector where ``master`` found this many."""
        if self.sector_class(master_source_count) == "dense":
            return self.dense_maximum_ratio
        return self.sparse_maximum_ratio

maximum_ratio(master_source_count)

Return the limit for a sector where master found this many.

Source code in src/hebog/validation/contracts.py

def maximum_ratio(self, master_source_count: int) -> float:
    """Return the limit for a sector where ``master`` found this many."""
    if self.sector_class(master_source_count) == "dense":
        return self.dense_maximum_ratio
    return self.sparse_maximum_ratio

sector_class(master_source_count)

Return the class of a sector where master found this many.

Source code in src/hebog/validation/contracts.py

def sector_class(self, master_source_count: int) -> SectorClass:
    """Return the class of a sector where ``master`` found this many."""
    if master_source_count >= self.dense_minimum_sources:
        return "dense"
    return "sparse"

ResourceProfile

Bases: _ContractModel

Representative admitted resources on one production node.

Source code in src/hebog/validation/contracts.py

class ResourceProfile(_ContractModel):
    """Representative admitted resources on one production node."""

    node_memory_bytes: int = Field(ge=1)
    workers_per_node: int = Field(ge=1)
    threads_per_worker: int = Field(ge=1)
    os_scheduler_headroom_bytes: int = Field(ge=0)
    concurrent_pipeline_reserve_bytes: int = Field(ge=0)
    worker_memory_limit_bytes: int = Field(ge=1)
    dask_target_fraction: float = Field(gt=0, lt=1)
    dask_spill_fraction: float = Field(gt=0, lt=1)
    dask_pause_fraction: float = Field(gt=0, lt=1)
    dask_terminate_fraction: float = Field(gt=0, le=1)
    spill_medium: Literal["worker-local-nvme"]

    @model_validator(mode="after")
    def validate_admission(self) -> Self:
        """Keep workers and Dask thresholds inside admitted memory."""
        admitted = (
            self.node_memory_bytes
            - self.os_scheduler_headroom_bytes
            - self.concurrent_pipeline_reserve_bytes
        )
        if admitted <= 0:
            raise ValueError("resource reserves must leave admitted memory")
        if self.workers_per_node * self.worker_memory_limit_bytes > admitted:
            raise ValueError("worker limits exceed admitted node memory")
        thresholds = (
            self.dask_target_fraction,
            self.dask_spill_fraction,
            self.dask_pause_fraction,
            self.dask_terminate_fraction,
        )
        if (
            tuple(sorted(thresholds)) != thresholds
            or len(set(thresholds)) != _DASK_MEMORY_THRESHOLD_COUNT
        ):
            raise ValueError(
                "Dask memory thresholds must be strictly increasing"
            )
        return self

validate_admission()

Keep workers and Dask thresholds inside admitted memory.

Source code in src/hebog/validation/contracts.py

@model_validator(mode="after")
def validate_admission(self) -> Self:
    """Keep workers and Dask thresholds inside admitted memory."""
    admitted = (
        self.node_memory_bytes
        - self.os_scheduler_headroom_bytes
        - self.concurrent_pipeline_reserve_bytes
    )
    if admitted <= 0:
        raise ValueError("resource reserves must leave admitted memory")
    if self.workers_per_node * self.worker_memory_limit_bytes > admitted:
        raise ValueError("worker limits exceed admitted node memory")
    thresholds = (
        self.dask_target_fraction,
        self.dask_spill_fraction,
        self.dask_pause_fraction,
        self.dask_terminate_fraction,
    )
    if (
        tuple(sorted(thresholds)) != thresholds
        or len(set(thresholds)) != _DASK_MEMORY_THRESHOLD_COUNT
    ):
        raise ValueError(
            "Dask memory thresholds must be strictly increasing"
        )
    return self

ScalabilityContract

Bases: _ContractModel

Frozen contract for the 100,000-square qualification case.

Source code in src/hebog/validation/contracts.py

class ScalabilityContract(_ContractModel):
    """Frozen contract for the 100,000-square qualification case."""

    schema_version: Literal[1]
    contract_id: str = Field(pattern=r"^[a-z0-9]+(?:-[a-z0-9]+)*$")
    status: Literal["frozen-provisional"]
    logical_shape_yx: tuple[int, int]
    input_planes: tuple[str, ...] = Field(min_length=1)
    output_planes: tuple[str, ...] = Field(min_length=1)
    maximum_live_float32_plane_equivalents: int = Field(ge=1)
    storage: StorageContract
    resource_profile: ResourceProfile
    candidate_tile_core_sizes: tuple[int, ...] = Field(min_length=1)
    maximum_halo_fraction_of_core: float = Field(gt=0, lt=0.5)
    maximum_worker_peak_fraction: float = Field(gt=0, lt=1)
    maximum_graph_tasks: int = Field(ge=1)
    node_gates: tuple[NodeScalingGate, ...]
    tile_batch_policy: str = Field(min_length=1)

    @model_validator(mode="after")
    def validate_contract(self) -> Self:
        """Protect the required geometry, topology matrix, and plane names."""
        if self.logical_shape_yx != (_MAXIMUM_MATRIX_SIZE,) * 2:
            raise ValueError(
                "scalability contract must describe 100000 square"
            )
        if len(set(self.input_planes)) != len(self.input_planes):
            raise ValueError("input plane names must be unique")
        if len(set(self.output_planes)) != len(self.output_planes):
            raise ValueError("output plane names must be unique")
        if tuple(sorted(set(self.candidate_tile_core_sizes))) != (
            self.candidate_tile_core_sizes
        ):
            raise ValueError(
                "candidate tile sizes must be unique and increasing"
            )
        nodes = tuple(gate.worker_nodes for gate in self.node_gates)
        if nodes != (1, 10, 50, 100, 200):
            raise ValueError("node gates must cover 1, 10, 50, 100, and 200")
        if self.node_gates[0].minimum_strong_scaling_efficiency is not None:
            raise ValueError(
                "one-node strong-scaling efficiency is the baseline"
            )
        return self

validate_contract()

Protect the required geometry, topology matrix, and plane names.

Source code in src/hebog/validation/contracts.py

@model_validator(mode="after")
def validate_contract(self) -> Self:
    """Protect the required geometry, topology matrix, and plane names."""
    if self.logical_shape_yx != (_MAXIMUM_MATRIX_SIZE,) * 2:
        raise ValueError(
            "scalability contract must describe 100000 square"
        )
    if len(set(self.input_planes)) != len(self.input_planes):
        raise ValueError("input plane names must be unique")
    if len(set(self.output_planes)) != len(self.output_planes):
        raise ValueError("output plane names must be unique")
    if tuple(sorted(set(self.candidate_tile_core_sizes))) != (
        self.candidate_tile_core_sizes
    ):
        raise ValueError(
            "candidate tile sizes must be unique and increasing"
        )
    nodes = tuple(gate.worker_nodes for gate in self.node_gates)
    if nodes != (1, 10, 50, 100, 200):
        raise ValueError("node gates must cover 1, 10, 50, 100, and 200")
    if self.node_gates[0].minimum_strong_scaling_efficiency is not None:
        raise ValueError(
            "one-node strong-scaling efficiency is the baseline"
        )
    return self

StorageContract

Bases: _ContractModel

Storage capabilities required by the extreme qualification case.

Source code in src/hebog/validation/contracts.py

class StorageContract(_ContractModel):
    """Storage capabilities required by the extreme qualification case."""

    input: Literal["window-readable-fits-or-chunk-addressable"]
    intermediate: Literal["independently-retryable-chunks"]
    output: Literal["chunked-then-compatible-materialisation"]
    shared_storage_identifier: str = Field(min_length=1)
    normal_run_spill_fraction_limit: float = Field(ge=0, le=1)

load_performance_matrix(path)

Load and validate a performance-matrix contract.

Source code in src/hebog/validation/contracts.py

def load_performance_matrix(path: Path) -> PerformanceMatrixContract:
    """Load and validate a performance-matrix contract."""
    return PerformanceMatrixContract.model_validate_json(
        path.read_text(encoding="utf-8")
    )

load_phase_four_measurement_contract(path)

Load frozen Phase 4 compact measurement meanings.

Source code in src/hebog/validation/contracts.py

def load_phase_four_measurement_contract(
    path: Path,
) -> PhaseFourMeasurementContract:
    """Load frozen Phase 4 compact measurement meanings."""
    return PhaseFourMeasurementContract.model_validate_json(
        path.read_text(encoding="utf-8")
    )

load_phase_four_scientific_gates(path)

Load frozen Phase 4 catalogue and uncertainty margins.

Source code in src/hebog/validation/contracts.py

def load_phase_four_scientific_gates(
    path: Path,
) -> PhaseFourScientificGates:
    """Load frozen Phase 4 catalogue and uncertainty margins."""
    return PhaseFourScientificGates.model_validate_json(
        path.read_text(encoding="utf-8")
    )

load_phase_three_scientific_gates(path)

Load frozen Phase 3 foreground-sensitive scientific margins.

Source code in src/hebog/validation/contracts.py

def load_phase_three_scientific_gates(
    path: Path,
) -> PhaseThreeScientificGates:
    """Load frozen Phase 3 foreground-sensitive scientific margins."""
    return PhaseThreeScientificGates.model_validate_json(
        path.read_text(encoding="utf-8")
    )

load_public_behaviours(path)

Load and validate the public-behaviour manifest.

Source code in src/hebog/validation/contracts.py

def load_public_behaviours(path: Path) -> PublicBehaviourManifest:
    """Load and validate the public-behaviour manifest."""
    return PublicBehaviourManifest.model_validate_json(
        path.read_text(encoding="utf-8")
    )

load_scalability_contract(path)

Load and validate an extreme-image scalability contract.

Source code in src/hebog/validation/contracts.py

def load_scalability_contract(path: Path) -> ScalabilityContract:
    """Load and validate an extreme-image scalability contract."""
    return ScalabilityContract.model_validate_json(
        path.read_text(encoding="utf-8")
    )