climate_ref_core.regression

Regression-baseline primitives for diagnostic test cases.

This package holds the building blocks for the two-bundle regression model, where we capture and commit a small, sanitised committed bundle of artefacts for a test case. These committed bundles are tracked in git and form the regression baseline for the test case.

We also snapshot the native files persisted by the execution. These files are typically large and not always portable, so we keep them out of git and refer to them by their sha256 digest in the committed bundle. These data will be able to be fetched from an object store in CI and replayed locally for debugging.

COMMITTED_BUNDLE_FILES = ('series.json', 'diagnostic.json', 'output.json') module-attribute

The committed CMEC artefacts tracked in git.

Their digests are tracked in :attr:Manifest.committed.

SCHEMA_VERSION = 2 module-attribute

Current manifest schema version.

Action

Bases: Enum

The verification action the CI gate selects for a single test case.

Source code in packages/climate-ref-core/src/climate_ref_core/regression/gate.py

class Action(enum.Enum):
    """The verification action the CI gate selects for a single test case."""

    SKIP = "skip"
    """Nothing relevant changed, or the case is not under regression management."""

    REPLAY = "replay"
    """Replay the cached native baseline and compare to the committed bundle."""

    FAIL = "fail"
    """The change is not permissible (unauthorised baseline change or bad version)."""

FAIL = 'fail' class-attribute instance-attribute

The change is not permissible (unauthorised baseline change or bad version).

REPLAY = 'replay' class-attribute instance-attribute

Replay the cached native baseline and compare to the committed bundle.

SKIP = 'skip' class-attribute instance-attribute

Nothing relevant changed, or the case is not under regression management.

GateDecision

The gate's decision for one test case: an :class:Action and why.

Source code in packages/climate-ref-core/src/climate_ref_core/regression/gate.py

@frozen
class GateDecision:
    """The gate's decision for one test case: an :class:`Action` and why."""

    action: Action
    """The selected verification action."""

    reason: str
    """A human-readable explanation, surfaced in CI logs."""

action instance-attribute

The selected verification action.

reason instance-attribute

A human-readable explanation, surfaced in CI logs.

Manifest

The on-disk manifest for a test case regression bundle.

Serialised as manifest.json with stable key ordering and a trailing newline, so repeated dumps are byte-identical.

Source code in packages/climate-ref-core/src/climate_ref_core/regression/manifest.py

@frozen
class Manifest:
    """
    The on-disk manifest for a test case regression bundle.

    Serialised as ``manifest.json`` with stable key ordering and a trailing newline,
    so repeated dumps are byte-identical.
    """

    schema: int
    """Manifest schema version; equals :data:`SCHEMA_VERSION` for current manifests."""

    test_case_version: int
    """Monotonic, author-bumped version coupling the bundle to its native outputs."""

    diagnostic_version: int
    """Author-declared ``Diagnostic.version`` at mint time.

    Monotonic. The gate fails a case whose in-code ``Diagnostic.version`` exceeds this value
    (a stale baseline that must be re-minted), and one whose in-code version drops below it.
    """

    committed: dict[str, str]
    """Digests of committed regression JSON artefacts: ``{relpath: sha256}``."""

    native: dict[str, NativeEntry]
    """Digests of curated native output files: ``{relpath: NativeEntry}``."""

    catalog_hash: str | None = None
    """Hash of the test case input ``catalog.yaml`` (its ``_metadata.hash``) that produced this baseline.
    This couples the baseline to its inputs.
    The CI gate fails a case whose live catalog hash no longer matches this value.
    """

    @classmethod
    def load(cls, path: Path) -> Manifest:
        """
        Load a manifest from ``manifest.json``.

        Parameters
        ----------
        path
            Path to the manifest file.

        Returns
        -------
        :
            The parsed manifest.

        Raises
        ------
        ValueError
            If the manifest is missing required keys or has malformed native entries
            (e.g. hand-edited or written by an incompatible version).
        """
        return cls.loads(path.read_text(encoding="utf-8"), source=str(path))

    @classmethod
    def loads(cls, text: str, *, source: str = "<string>") -> Manifest:
        """
        Parse a manifest from its JSON text.

        Used when the manifest does not live on disk at parse time,
        e.g. when reading the base-branch copy via ``git show`` for the CI coupling gate.

        Parameters
        ----------
        text
            The manifest JSON.
        source
            A label for the text's origin, used in error messages.

        Returns
        -------
        :
            The parsed manifest.

        Raises
        ------
        ValueError
            If the manifest is missing required keys or has malformed native entries
            (e.g. hand-edited or written by an incompatible version).
        """
        data = json.loads(text)
        missing = [
            key
            for key in ("schema", "test_case_version", "diagnostic_version", "committed", "native")
            if key not in data
        ]
        if missing:
            raise ValueError(
                f"Invalid manifest {source}: missing required keys {missing}. "
                "The manifest may be corrupted or written by an incompatible version; "
                "regenerate it with `ref test-cases run --force-regen`."
            )
        schema = data["schema"]
        if isinstance(schema, bool) or not isinstance(schema, int) or schema != SCHEMA_VERSION:
            raise ValueError(
                f"Invalid manifest {source}: unsupported schema {schema!r}, "
                f"expected {SCHEMA_VERSION}. The manifest was written by an incompatible "
                "version; regenerate it with `ref test-cases run --force-regen`."
            )
        diagnostic_version = data["diagnostic_version"]
        if (
            isinstance(diagnostic_version, bool)
            or not isinstance(diagnostic_version, int)
            or diagnostic_version < 1
        ):
            raise ValueError(
                f"Invalid manifest {source}: diagnostic_version {diagnostic_version!r} "
                "is invalid; expected an integer >= 1."
            )
        try:
            native = {
                relpath: NativeEntry(sha256=entry["sha256"], size=entry["size"])
                for relpath, entry in data["native"].items()
            }
        except (KeyError, TypeError, AttributeError) as exc:
            raise ValueError(
                f"Invalid manifest {source}: malformed 'native' entry ({exc!r}). "
                "Each entry must be a mapping with 'sha256' and 'size' keys."
            ) from exc
        # Reject hand-edited or hostile manifests that could escape the
        # destination directory or carry a malformed digest when materialised.
        for relpath, entry in native.items():
            try:
                safe_path(relpath, label="native path")
            except ValueError as exc:
                raise ValueError(f"Invalid manifest {source}: {exc}") from exc
            _validate_digest(entry.sha256)
            if isinstance(entry.size, bool) or not isinstance(entry.size, int) or entry.size < 0:
                raise ValueError(
                    f"Invalid manifest {source}: native entry {relpath!r} has invalid size "
                    f"{entry.size!r}; expected a non-negative integer."
                )
        return cls(
            schema=data["schema"],
            test_case_version=data["test_case_version"],
            diagnostic_version=diagnostic_version,
            committed=dict(data["committed"]),
            native=native,
            # TODO: remove optonality when all manifests have this field.
            catalog_hash=data.get("catalog_hash"),
        )

    def dump(self, path: Path) -> None:
        """
        Write the manifest to ``manifest.json``.

        Keys are stably ordered (``sort_keys=True``) and a trailing newline is added,
        so ``dump`` followed by ``load`` round-trips byte-identically.

        Parameters
        ----------
        path
            Path to write the manifest to.
        """
        payload = {
            "schema": self.schema,
            "test_case_version": self.test_case_version,
            "diagnostic_version": self.diagnostic_version,
            "catalog_hash": self.catalog_hash,
            "committed": self.committed,
            "native": {relpath: asdict(entry) for relpath, entry in self.native.items()},
        }
        text = json.dumps(payload, indent=2, sort_keys=True) + "\n"
        path.write_text(text, encoding="utf-8")

catalog_hash = None class-attribute instance-attribute

Hash of the test case input catalog.yaml (its _metadata.hash) that produced this baseline. This couples the baseline to its inputs. The CI gate fails a case whose live catalog hash no longer matches this value.

committed instance-attribute

Digests of committed regression JSON artefacts: {relpath: sha256}.

diagnostic_version instance-attribute

Author-declared Diagnostic.version at mint time.

Monotonic. The gate fails a case whose in-code Diagnostic.version exceeds this value (a stale baseline that must be re-minted), and one whose in-code version drops below it.

native instance-attribute

Digests of curated native output files: {relpath: NativeEntry}.

schema instance-attribute

Manifest schema version; equals :data:SCHEMA_VERSION for current manifests.

test_case_version instance-attribute

Monotonic, author-bumped version coupling the bundle to its native outputs.

dump(path)

Write the manifest to manifest.json.

Keys are stably ordered (sort_keys=True) and a trailing newline is added, so dump followed by load round-trips byte-identically.

Parameters:

Name	Type	Description	Default
path	Path	Path to write the manifest to.	required

Source code in packages/climate-ref-core/src/climate_ref_core/regression/manifest.py

def dump(self, path: Path) -> None:
    """
    Write the manifest to ``manifest.json``.

    Keys are stably ordered (``sort_keys=True``) and a trailing newline is added,
    so ``dump`` followed by ``load`` round-trips byte-identically.

    Parameters
    ----------
    path
        Path to write the manifest to.
    """
    payload = {
        "schema": self.schema,
        "test_case_version": self.test_case_version,
        "diagnostic_version": self.diagnostic_version,
        "catalog_hash": self.catalog_hash,
        "committed": self.committed,
        "native": {relpath: asdict(entry) for relpath, entry in self.native.items()},
    }
    text = json.dumps(payload, indent=2, sort_keys=True) + "\n"
    path.write_text(text, encoding="utf-8")

load(path) classmethod

Load a manifest from manifest.json.

Parameters:

Name	Type	Description	Default
path	Path	Path to the manifest file.	required

Returns:

Type	Description
Manifest	The parsed manifest.

Raises:

Type	Description
ValueError	If the manifest is missing required keys or has malformed native entries (e.g. hand-edited or written by an incompatible version).

Source code in packages/climate-ref-core/src/climate_ref_core/regression/manifest.py

@classmethod
def load(cls, path: Path) -> Manifest:
    """
    Load a manifest from ``manifest.json``.

    Parameters
    ----------
    path
        Path to the manifest file.

    Returns
    -------
    :
        The parsed manifest.

    Raises
    ------
    ValueError
        If the manifest is missing required keys or has malformed native entries
        (e.g. hand-edited or written by an incompatible version).
    """
    return cls.loads(path.read_text(encoding="utf-8"), source=str(path))

loads(text, *, source='<string>') classmethod

Parse a manifest from its JSON text.

Used when the manifest does not live on disk at parse time, e.g. when reading the base-branch copy via git show for the CI coupling gate.

Parameters:

Name	Type	Description	Default
text	str	The manifest JSON.	required
source	str	A label for the text's origin, used in error messages.	'<string>'

Returns:

Type	Description
Manifest	The parsed manifest.

Raises:

Type	Description
ValueError	If the manifest is missing required keys or has malformed native entries (e.g. hand-edited or written by an incompatible version).

Source code in packages/climate-ref-core/src/climate_ref_core/regression/manifest.py

@classmethod
def loads(cls, text: str, *, source: str = "<string>") -> Manifest:
    """
    Parse a manifest from its JSON text.

    Used when the manifest does not live on disk at parse time,
    e.g. when reading the base-branch copy via ``git show`` for the CI coupling gate.

    Parameters
    ----------
    text
        The manifest JSON.
    source
        A label for the text's origin, used in error messages.

    Returns
    -------
    :
        The parsed manifest.

    Raises
    ------
    ValueError
        If the manifest is missing required keys or has malformed native entries
        (e.g. hand-edited or written by an incompatible version).
    """
    data = json.loads(text)
    missing = [
        key
        for key in ("schema", "test_case_version", "diagnostic_version", "committed", "native")
        if key not in data
    ]
    if missing:
        raise ValueError(
            f"Invalid manifest {source}: missing required keys {missing}. "
            "The manifest may be corrupted or written by an incompatible version; "
            "regenerate it with `ref test-cases run --force-regen`."
        )
    schema = data["schema"]
    if isinstance(schema, bool) or not isinstance(schema, int) or schema != SCHEMA_VERSION:
        raise ValueError(
            f"Invalid manifest {source}: unsupported schema {schema!r}, "
            f"expected {SCHEMA_VERSION}. The manifest was written by an incompatible "
            "version; regenerate it with `ref test-cases run --force-regen`."
        )
    diagnostic_version = data["diagnostic_version"]
    if (
        isinstance(diagnostic_version, bool)
        or not isinstance(diagnostic_version, int)
        or diagnostic_version < 1
    ):
        raise ValueError(
            f"Invalid manifest {source}: diagnostic_version {diagnostic_version!r} "
            "is invalid; expected an integer >= 1."
        )
    try:
        native = {
            relpath: NativeEntry(sha256=entry["sha256"], size=entry["size"])
            for relpath, entry in data["native"].items()
        }
    except (KeyError, TypeError, AttributeError) as exc:
        raise ValueError(
            f"Invalid manifest {source}: malformed 'native' entry ({exc!r}). "
            "Each entry must be a mapping with 'sha256' and 'size' keys."
        ) from exc
    # Reject hand-edited or hostile manifests that could escape the
    # destination directory or carry a malformed digest when materialised.
    for relpath, entry in native.items():
        try:
            safe_path(relpath, label="native path")
        except ValueError as exc:
            raise ValueError(f"Invalid manifest {source}: {exc}") from exc
        _validate_digest(entry.sha256)
        if isinstance(entry.size, bool) or not isinstance(entry.size, int) or entry.size < 0:
            raise ValueError(
                f"Invalid manifest {source}: native entry {relpath!r} has invalid size "
                f"{entry.size!r}; expected a non-negative integer."
            )
    return cls(
        schema=data["schema"],
        test_case_version=data["test_case_version"],
        diagnostic_version=diagnostic_version,
        committed=dict(data["committed"]),
        native=native,
        # TODO: remove optonality when all manifests have this field.
        catalog_hash=data.get("catalog_hash"),
    )

NativeEntry

A single curated native output file recorded in the manifest.

Source code in packages/climate-ref-core/src/climate_ref_core/regression/manifest.py

@frozen
class NativeEntry:
    """A single curated native output file recorded in the manifest."""

    sha256: str
    """Hex-encoded sha256 digest of the curated file."""

    size: int
    """Size of the curated file in bytes."""

sha256 instance-attribute

Hex-encoded sha256 digest of the curated file.

size instance-attribute

Size of the curated file in bytes.

NativeStore

Content-addressed store for native baseline blobs, keyed by sha256 hex digest.

A local store (a file:// URL or a bare path) always reads and writes without credentials. A remote store reads anonymously through :mod:pooch, and writes only when it is given an :class:S3WriteConfig.

Parameters:

Name	Type	Description	Default
url		Where blobs are read from: an http(s):// base URL (no trailing slash needed), a file:///absolute/path URL, or a bare filesystem path.	required
cache_dir		Local directory pooch caches anonymous remote downloads in. A local store does not use it.	required
write		The S3 endpoint, bucket and credentials a remote store writes with. None leaves a remote store read-only. A local store ignores it and is always writable.	required

Source code in packages/climate-ref-core/src/climate_ref_core/regression/store.py

@frozen
class NativeStore:
    """
    Content-addressed store for native baseline blobs, keyed by sha256 hex digest.

    A local store (a ``file://`` URL or a bare path) always reads and writes without credentials.
    A remote store reads anonymously through :mod:`pooch`,
    and writes only when it is given an :class:`S3WriteConfig`.

    Parameters
    ----------
    url
        Where blobs are read from: an ``http(s)://`` base URL (no trailing slash needed),
        a ``file:///absolute/path`` URL, or a bare filesystem path.
    cache_dir
        Local directory pooch caches anonymous remote downloads in.
        A local store does not use it.
    write
        The S3 endpoint, bucket and credentials a remote store writes with.
        ``None`` leaves a remote store read-only.
        A local store ignores it and is always writable.
    """

    url: str
    cache_dir: Path | None = None
    write: S3WriteConfig | None = None
    root: Path | None = field(init=False)
    """The local store root, or ``None`` when this store is remote."""

    @root.default
    def _resolve_root(self) -> Path | None:
        """Resolve the local root once at construction, which also validates the URL."""
        return _local_root(self.url)

    def _blob_path(self, digest: str, root: Path) -> Path:
        """
        Return the canonical on-disk path for a blob in a local store.

        The digest is validated as 64-character lowercase hex first,
        so a malformed or hostile digest cannot be used to construct a path outside the root.
        """
        _validate_digest(digest)
        return root / digest[:2] / digest

    def _pooch_cache(self) -> Path:
        """Return the directory pooch caches downloads in, which an anonymous read needs."""
        if self.cache_dir is None:
            raise ValueError(f"Reading from {self.url} needs a cache directory, but none is configured.")
        return self.cache_dir

    def has(self, digest: str) -> bool:
        """
        Return ``True`` if the blob identified by ``digest`` is available.

        A local store checks its canonical path,
        a writable remote store ``HEAD``s the object,
        and an anonymous remote store checks the pooch cache.

        Parameters
        ----------
        digest
            The sha256 hex digest of the blob.

        Returns
        -------
        :
            ``True`` when the blob is present, ``False`` when it is not.
        """
        _validate_digest(digest)
        root = self.root
        if root is not None:
            return self._blob_path(digest, root).exists()
        write = self.write
        if write is None:
            return (self._pooch_cache() / digest).exists()

        from botocore.exceptions import ClientError  # noqa: PLC0415 - optional dependency

        try:
            write.client().head_object(Bucket=write.bucket, Key=digest)
        except ClientError as exc:
            if _is_missing(exc):
                return False
            raise
        return True

    def fetch(self, digest: str, dest: Path) -> None:
        """
        Fetch the blob identified by ``digest`` and write it to ``dest``.

        The sha256 of the written file is verified to equal ``digest``.

        Parameters
        ----------
        digest
            The sha256 hex digest of the blob to fetch.
        dest
            Destination path to write the blob to.
            Parent directories are created if they do not exist.

        Raises
        ------
        FileNotFoundError
            If the blob is not present in the store.
        ValueError
            If the fetched blob's sha256 does not match ``digest``.
        """
        _validate_digest(digest)
        root = self.root
        dest.parent.mkdir(parents=True, exist_ok=True)
        if root is not None:
            blob = self._blob_path(digest, root)
            if not blob.exists():
                raise FileNotFoundError(f"Blob {digest!r} not found in local store at {root}")
            shutil.copy2(str(blob), str(dest))
        elif self.write is None:
            registry = _pooch_manager(self.url.rstrip("/"), str(self._pooch_cache()))
            registry.registry[digest] = digest  # content-addressed: hash == name
            shutil.copy2(registry.fetch(digest), str(dest))
        else:
            from botocore.exceptions import ClientError  # noqa: PLC0415 - optional dependency

            write = self.write
            try:
                write.client().download_file(write.bucket, digest, str(dest))
            except ClientError as exc:
                if _is_missing(exc):
                    raise FileNotFoundError(
                        f"Blob {digest!r} not found in R2 bucket {write.bucket!r}"
                    ) from exc
                raise
        _verify_hash_matches(dest, digest)
        logger.debug(f"NativeStore.fetch: {digest} -> {dest}")

    def put(self, path: Path) -> str:
        """
        Store the file at ``path`` and return its sha256 hex digest.

        The blob is content-addressed, so storing one that is already present is skipped and
        re-minting is cheap.

        Parameters
        ----------
        path
            Path to the file to store.

        Returns
        -------
        :
            The sha256 hex digest of the stored blob.

        Raises
        ------
        NotImplementedError
            If this is an anonymous remote store, which cannot write.
        """
        digest = sha256_file(path)
        root = self.root
        if root is not None:
            blob = self._blob_path(digest, root)
            if blob.exists():
                logger.debug(f"NativeStore.put: {digest} already present, skipping copy")
                return digest
            blob.parent.mkdir(parents=True, exist_ok=True)
            shutil.copy2(str(path), str(blob))
        else:
            write = self.write
            if write is None:
                raise NotImplementedError(
                    f"Native store {self.url} is a public-read store, so put() is not supported. "
                    "Mint against a local path or a credentialed remote store."
                )
            if self.has(digest):
                logger.debug(f"NativeStore.put: {digest} already present, skipping upload")
                return digest
            write.client().upload_file(str(path), write.bucket, digest)
        logger.debug(f"NativeStore.put: {path} -> {digest}")
        return digest

    def preflight(self) -> None:
        """
        Verify the store is reachable and usable before relying on it.

        A local store's root is created if needed and checked for writability.
        A writable remote store performs a cheap authenticated ``HEAD`` on a sentinel key,
        which is expected to be absent:
        a ``404`` means the request authenticated and the store is usable,
        while ``401`` / ``403`` become actionable errors,
        so a misconfigured credential is caught before the (slow) diagnostic run rather than after.
        ``head_object`` is used rather than ``head_bucket``,
        so the check works with least-privilege, object-scoped tokens.

        An anonymous remote store has nothing to verify up front.
        It has no credentials, and every read is hash-checked per blob.

        Raises
        ------
        NativeStoreUnavailableError
            If the store cannot be reached or used, with an operator-facing message.
        """
        root = self.root
        if root is not None:
            try:
                root.mkdir(parents=True, exist_ok=True)
            except OSError as exc:
                raise NativeStoreUnavailableError(
                    f"Local native store root {root} could not be created: {exc}"
                ) from exc
            if not os.access(root, os.W_OK):
                raise NativeStoreUnavailableError(f"Local native store root {root} is not writable.")
            logger.debug(f"Local native store ready at {root}")
            return
        write = self.write
        if write is None:
            return

        from botocore.exceptions import ClientError  # noqa: PLC0415 - optional dependency

        try:
            write.client().head_object(Bucket=write.bucket, Key=_PREFLIGHT_PROBE_KEY)
        except ClientError as exc:
            status = _http_status(exc)
            if status == _HTTP_NOT_FOUND:
                pass  # authenticated, and the probe object is simply absent, so the store is usable
            elif status in _AUTH_REJECTED_STATUSES:
                raise NativeStoreUnavailableError(
                    f"Native store authentication failed (HTTP {status}) for bucket {write.bucket!r} at "
                    f"{write.endpoint_url}: the credentials were rejected or malformed. Check "
                    f"REF_NATIVE_STORE_PROFILE, or REF_NATIVE_STORE_ACCESS_KEY_ID / "
                    f"REF_NATIVE_STORE_SECRET_ACCESS_KEY."
                ) from exc
            elif status == _HTTP_FORBIDDEN:
                raise NativeStoreUnavailableError(
                    f"Native store access denied (HTTP 403) for bucket {write.bucket!r} at "
                    f"{write.endpoint_url}: the request was forbidden. The secret key may be wrong, "
                    f"or the token may lack object read and write on this bucket. Check the "
                    f"credentials and the token's permissions."
                ) from exc
            else:
                raise NativeStoreUnavailableError(
                    f"Native store preflight failed (HTTP {status}) for bucket {write.bucket!r} at "
                    f"{write.endpoint_url}: {exc}"
                ) from exc
        logger.info(f"Native store authenticated: bucket {write.bucket!r} at {write.endpoint_url}")

root = field(init=False) class-attribute instance-attribute

The local store root, or None when this store is remote.

fetch(digest, dest)

Fetch the blob identified by digest and write it to dest.

The sha256 of the written file is verified to equal digest.

Parameters:

Name	Type	Description	Default
digest	str	The sha256 hex digest of the blob to fetch.	required
dest	Path	Destination path to write the blob to. Parent directories are created if they do not exist.	required

Raises:

Type	Description
FileNotFoundError	If the blob is not present in the store.
ValueError	If the fetched blob's sha256 does not match digest.

Source code in packages/climate-ref-core/src/climate_ref_core/regression/store.py

def fetch(self, digest: str, dest: Path) -> None:
    """
    Fetch the blob identified by ``digest`` and write it to ``dest``.

    The sha256 of the written file is verified to equal ``digest``.

    Parameters
    ----------
    digest
        The sha256 hex digest of the blob to fetch.
    dest
        Destination path to write the blob to.
        Parent directories are created if they do not exist.

    Raises
    ------
    FileNotFoundError
        If the blob is not present in the store.
    ValueError
        If the fetched blob's sha256 does not match ``digest``.
    """
    _validate_digest(digest)
    root = self.root
    dest.parent.mkdir(parents=True, exist_ok=True)
    if root is not None:
        blob = self._blob_path(digest, root)
        if not blob.exists():
            raise FileNotFoundError(f"Blob {digest!r} not found in local store at {root}")
        shutil.copy2(str(blob), str(dest))
    elif self.write is None:
        registry = _pooch_manager(self.url.rstrip("/"), str(self._pooch_cache()))
        registry.registry[digest] = digest  # content-addressed: hash == name
        shutil.copy2(registry.fetch(digest), str(dest))
    else:
        from botocore.exceptions import ClientError  # noqa: PLC0415 - optional dependency

        write = self.write
        try:
            write.client().download_file(write.bucket, digest, str(dest))
        except ClientError as exc:
            if _is_missing(exc):
                raise FileNotFoundError(
                    f"Blob {digest!r} not found in R2 bucket {write.bucket!r}"
                ) from exc
            raise
    _verify_hash_matches(dest, digest)
    logger.debug(f"NativeStore.fetch: {digest} -> {dest}")

has(digest)

Return True if the blob identified by digest is available.

A local store checks its canonical path, a writable remote store HEADs the object, and an anonymous remote store checks the pooch cache.

Parameters:

Name	Type	Description	Default
digest	str	The sha256 hex digest of the blob.	required

Returns:

Type	Description
bool	True when the blob is present, False when it is not.

Source code in packages/climate-ref-core/src/climate_ref_core/regression/store.py

def has(self, digest: str) -> bool:
    """
    Return ``True`` if the blob identified by ``digest`` is available.

    A local store checks its canonical path,
    a writable remote store ``HEAD``s the object,
    and an anonymous remote store checks the pooch cache.

    Parameters
    ----------
    digest
        The sha256 hex digest of the blob.

    Returns
    -------
    :
        ``True`` when the blob is present, ``False`` when it is not.
    """
    _validate_digest(digest)
    root = self.root
    if root is not None:
        return self._blob_path(digest, root).exists()
    write = self.write
    if write is None:
        return (self._pooch_cache() / digest).exists()

    from botocore.exceptions import ClientError  # noqa: PLC0415 - optional dependency

    try:
        write.client().head_object(Bucket=write.bucket, Key=digest)
    except ClientError as exc:
        if _is_missing(exc):
            return False
        raise
    return True

preflight()

Verify the store is reachable and usable before relying on it.

A local store's root is created if needed and checked for writability. A writable remote store performs a cheap authenticated HEAD on a sentinel key, which is expected to be absent: a 404 means the request authenticated and the store is usable, while 401 / 403 become actionable errors, so a misconfigured credential is caught before the (slow) diagnostic run rather than after. head_object is used rather than head_bucket, so the check works with least-privilege, object-scoped tokens.

An anonymous remote store has nothing to verify up front. It has no credentials, and every read is hash-checked per blob.

Raises:

Type	Description
NativeStoreUnavailableError	If the store cannot be reached or used, with an operator-facing message.

Source code in packages/climate-ref-core/src/climate_ref_core/regression/store.py

def preflight(self) -> None:
    """
    Verify the store is reachable and usable before relying on it.

    A local store's root is created if needed and checked for writability.
    A writable remote store performs a cheap authenticated ``HEAD`` on a sentinel key,
    which is expected to be absent:
    a ``404`` means the request authenticated and the store is usable,
    while ``401`` / ``403`` become actionable errors,
    so a misconfigured credential is caught before the (slow) diagnostic run rather than after.
    ``head_object`` is used rather than ``head_bucket``,
    so the check works with least-privilege, object-scoped tokens.

    An anonymous remote store has nothing to verify up front.
    It has no credentials, and every read is hash-checked per blob.

    Raises
    ------
    NativeStoreUnavailableError
        If the store cannot be reached or used, with an operator-facing message.
    """
    root = self.root
    if root is not None:
        try:
            root.mkdir(parents=True, exist_ok=True)
        except OSError as exc:
            raise NativeStoreUnavailableError(
                f"Local native store root {root} could not be created: {exc}"
            ) from exc
        if not os.access(root, os.W_OK):
            raise NativeStoreUnavailableError(f"Local native store root {root} is not writable.")
        logger.debug(f"Local native store ready at {root}")
        return
    write = self.write
    if write is None:
        return

    from botocore.exceptions import ClientError  # noqa: PLC0415 - optional dependency

    try:
        write.client().head_object(Bucket=write.bucket, Key=_PREFLIGHT_PROBE_KEY)
    except ClientError as exc:
        status = _http_status(exc)
        if status == _HTTP_NOT_FOUND:
            pass  # authenticated, and the probe object is simply absent, so the store is usable
        elif status in _AUTH_REJECTED_STATUSES:
            raise NativeStoreUnavailableError(
                f"Native store authentication failed (HTTP {status}) for bucket {write.bucket!r} at "
                f"{write.endpoint_url}: the credentials were rejected or malformed. Check "
                f"REF_NATIVE_STORE_PROFILE, or REF_NATIVE_STORE_ACCESS_KEY_ID / "
                f"REF_NATIVE_STORE_SECRET_ACCESS_KEY."
            ) from exc
        elif status == _HTTP_FORBIDDEN:
            raise NativeStoreUnavailableError(
                f"Native store access denied (HTTP 403) for bucket {write.bucket!r} at "
                f"{write.endpoint_url}: the request was forbidden. The secret key may be wrong, "
                f"or the token may lack object read and write on this bucket. Check the "
                f"credentials and the token's permissions."
            ) from exc
        else:
            raise NativeStoreUnavailableError(
                f"Native store preflight failed (HTTP {status}) for bucket {write.bucket!r} at "
                f"{write.endpoint_url}: {exc}"
            ) from exc
    logger.info(f"Native store authenticated: bucket {write.bucket!r} at {write.endpoint_url}")

put(path)

Store the file at path and return its sha256 hex digest.

The blob is content-addressed, so storing one that is already present is skipped and re-minting is cheap.

Parameters:

Name	Type	Description	Default
path	Path	Path to the file to store.	required

Returns:

Type	Description
str	The sha256 hex digest of the stored blob.

Raises:

Type	Description
NotImplementedError	If this is an anonymous remote store, which cannot write.

Source code in packages/climate-ref-core/src/climate_ref_core/regression/store.py

def put(self, path: Path) -> str:
    """
    Store the file at ``path`` and return its sha256 hex digest.

    The blob is content-addressed, so storing one that is already present is skipped and
    re-minting is cheap.

    Parameters
    ----------
    path
        Path to the file to store.

    Returns
    -------
    :
        The sha256 hex digest of the stored blob.

    Raises
    ------
    NotImplementedError
        If this is an anonymous remote store, which cannot write.
    """
    digest = sha256_file(path)
    root = self.root
    if root is not None:
        blob = self._blob_path(digest, root)
        if blob.exists():
            logger.debug(f"NativeStore.put: {digest} already present, skipping copy")
            return digest
        blob.parent.mkdir(parents=True, exist_ok=True)
        shutil.copy2(str(path), str(blob))
    else:
        write = self.write
        if write is None:
            raise NotImplementedError(
                f"Native store {self.url} is a public-read store, so put() is not supported. "
                "Mint against a local path or a credentialed remote store."
            )
        if self.has(digest):
            logger.debug(f"NativeStore.put: {digest} already present, skipping upload")
            return digest
        write.client().upload_file(str(path), write.bucket, digest)
    logger.debug(f"NativeStore.put: {path} -> {digest}")
    return digest

NativeStoreUnavailableError

Bases: RuntimeError

Raised when a native store cannot be reached or used.

Covers rejected credentials, a missing bucket, or an unwritable local directory. The message is operator-facing and actionable (it names the env vars / path to check), so callers can surface it directly.

Source code in packages/climate-ref-core/src/climate_ref_core/regression/store.py

class NativeStoreUnavailableError(RuntimeError):
    """
    Raised when a native store cannot be reached or used.

    Covers rejected credentials, a missing bucket, or an unwritable local directory.
    The message is operator-facing and actionable (it names the env vars / path to check),
    so callers can surface it directly.
    """

S3WriteConfig

Where a remote store writes, and how it authenticates.

Routing (endpoint_url and bucket) is non-secret and comes from the application config. Credentials are resolved from the environment at build time and are never persisted.

Parameters:

Name	Type	Description	Default
endpoint_url		S3 API endpoint for the bucket's account, without the bucket (e.g. https://<account>.eu.r2.cloudflarestorage.com).	required
bucket		Name of the R2 bucket (e.g. ref-baselines).	required
access_key_id		R2 access-key id, or "" to fall through to profile / boto3's default chain.	required
secret_access_key		R2 secret-access-key, or "" to fall through to profile / boto3's default chain.	required
profile		Named AWS/R2 profile to authenticate with, or "" for the default session. Ignored when explicit access_key_id / secret_access_key are supplied.	required

Source code in packages/climate-ref-core/src/climate_ref_core/regression/store.py

@frozen
class S3WriteConfig:
    """
    Where a remote store writes, and how it authenticates.

    Routing (``endpoint_url`` and ``bucket``) is non-secret and comes from the application config.
    Credentials are resolved from the environment at build time and are never persisted.

    Parameters
    ----------
    endpoint_url
        S3 API endpoint for the bucket's account, without the bucket
        (e.g. ``https://<account>.eu.r2.cloudflarestorage.com``).
    bucket
        Name of the R2 bucket (e.g. ``ref-baselines``).
    access_key_id
        R2 access-key id, or ``""`` to fall through to ``profile`` / boto3's default chain.
    secret_access_key
        R2 secret-access-key, or ``""`` to fall through to ``profile`` / boto3's default chain.
    profile
        Named AWS/R2 profile to authenticate with, or ``""`` for the default session.
        Ignored when explicit ``access_key_id`` / ``secret_access_key`` are supplied.
    """

    endpoint_url: str
    bucket: str
    access_key_id: str = field(default="", repr=False)
    secret_access_key: str = field(default="", repr=False)
    profile: str = ""

    def __attrs_post_init__(self) -> None:
        """Fail fast at construction (mint startup) when the routing config is missing."""
        if not self.endpoint_url:
            raise ValueError(
                "R2 native store requires an S3 endpoint URL. Set REF_NATIVE_STORE_S3_ENDPOINT_URL "
                "(e.g. https://<account>.eu.r2.cloudflarestorage.com)."
            )
        if not self.bucket:
            raise ValueError(
                "R2 native store requires a bucket name. Set REF_NATIVE_STORE_BUCKET (e.g. ref-baselines)."
            )

    def client(self) -> Any:
        """Return the cached boto3 S3 client for this endpoint and these credentials."""
        return _s3_client(self.endpoint_url, self.access_key_id, self.secret_access_key, self.profile)

__attrs_post_init__()

Fail fast at construction (mint startup) when the routing config is missing.

Source code in packages/climate-ref-core/src/climate_ref_core/regression/store.py

def __attrs_post_init__(self) -> None:
    """Fail fast at construction (mint startup) when the routing config is missing."""
    if not self.endpoint_url:
        raise ValueError(
            "R2 native store requires an S3 endpoint URL. Set REF_NATIVE_STORE_S3_ENDPOINT_URL "
            "(e.g. https://<account>.eu.r2.cloudflarestorage.com)."
        )
    if not self.bucket:
        raise ValueError(
            "R2 native store requires a bucket name. Set REF_NATIVE_STORE_BUCKET (e.g. ref-baselines)."
        )

client()

Return the cached boto3 S3 client for this endpoint and these credentials.

Source code in packages/climate-ref-core/src/climate_ref_core/regression/store.py

def client(self) -> Any:
    """Return the cached boto3 S3 client for this endpoint and these credentials."""
    return _s3_client(self.endpoint_url, self.access_key_id, self.secret_access_key, self.profile)

Tolerance

Float comparison tolerance for bundle regression checks.

Parameters:

Name	Type	Description	Default
rtol		Relative tolerance — the allowed proportional difference between expected and actual float values (default 1e-5).	required
atol		Absolute tolerance — the floor difference always permitted, regardless of magnitude (default 1e-8).	required

Source code in packages/climate-ref-core/src/climate_ref_core/regression/compare.py

@frozen
class Tolerance:
    """
    Float comparison tolerance for bundle regression checks.

    Parameters
    ----------
    rtol
        Relative tolerance — the allowed proportional difference between expected
        and actual float values (default ``1e-5``).
    atol
        Absolute tolerance — the floor difference always permitted,
        regardless of magnitude (default ``1e-8``).
    """

    rtol: float = 1e-5
    atol: float = 1e-8

assert_bundle_regression(expected_path, actual_path, *, slug, tol=Tolerance(), replacements)

Assert that a regenerated committed-bundle JSON file matches the committed copy.

Algorithm:

1. Byte-equal fast path — if the raw bytes of both files are identical, return immediately.
2. Parse both files as JSON.
3. Rewrite both dict keys and leaf string values in the regenerated (actual) document using replacements, applied longest-key-first so that a shorter placeholder cannot pre-empt a longer overlapping one.
4. Call :func:compare_json_content with the given tolerance.
5. Raise AssertionError with the full mismatch list and a remediation hint if any mismatches are found.

The replacements map follows the convention used throughout the testing infrastructure: keys are real runtime values (absolute paths, recipe-dir timestamps), values are the committed-bundle placeholders (e.g. "<OUTPUT_DIR>", "<TEST_DATA_DIR>", "<RECIPE_RUN>"). Only the actual document is rewritten: the committed expected file is assumed to already contain placeholders, which :func:~climate_ref_core.regression.capture.write_committed_bundle guarantees at capture time. A hand-edited baseline with raw paths will surface as ordinary value mismatches.

Both <OUTPUT_DIR> and <RECIPE_RUN> participate in dict-KEY rewriting because ESMValTool's output.json embeds absolute paths as object keys.

Parameters:

Name	Type	Description	Default
expected_path	Path	Path to the committed (on-disk) bundle file containing placeholders.	required
actual_path	Path	Path to the regenerated bundle file containing real runtime paths.	required
slug	str	Diagnostic slug used in error messages.	required
tol	Tolerance	Float comparison tolerance.	Tolerance()
replacements	dict[str, str]	Mapping of {real_value: placeholder} applied to the actual document.	required

Raises:

Type	Description
AssertionError	If the bundles differ beyond tolerance after sanitisation.

Notes

If expected_path does not exist (a legacy regression without a committed bundle), the comparison is skipped silently and the function returns.

Source code in packages/climate-ref-core/src/climate_ref_core/regression/compare.py

def assert_bundle_regression(
    expected_path: Path,
    actual_path: Path,
    *,
    slug: str,
    tol: Tolerance = Tolerance(),
    replacements: dict[str, str],
) -> None:
    """
    Assert that a regenerated committed-bundle JSON file matches the committed copy.

    Algorithm:

    1. **Byte-equal fast path** — if the raw bytes of both files are identical, return immediately.
    2. Parse both files as JSON.
    3. Rewrite both dict *keys* and leaf string *values* in the regenerated
       (``actual``) document using ``replacements``, applied **longest-key-first**
       so that a shorter placeholder cannot pre-empt a longer overlapping one.
    4. Call :func:`compare_json_content` with the given tolerance.
    5. Raise ``AssertionError`` with the full mismatch list and a remediation hint
       if any mismatches are found.

    The replacements map follows the convention used throughout the testing
    infrastructure: keys are real runtime values (absolute paths, recipe-dir
    timestamps), values are the committed-bundle placeholders (e.g.
    ``"<OUTPUT_DIR>"``, ``"<TEST_DATA_DIR>"``, ``"<RECIPE_RUN>"``).
    Only the *actual* document is rewritten: the committed *expected* file is
    assumed to already contain placeholders, which
    :func:`~climate_ref_core.regression.capture.write_committed_bundle` guarantees
    at capture time. A hand-edited baseline with raw paths will surface as ordinary
    value mismatches.

    Both ``<OUTPUT_DIR>`` and ``<RECIPE_RUN>`` participate in dict-KEY rewriting
    because ESMValTool's ``output.json`` embeds absolute paths as object keys.

    Parameters
    ----------
    expected_path
        Path to the committed (on-disk) bundle file containing placeholders.
    actual_path
        Path to the regenerated bundle file containing real runtime paths.
    slug
        Diagnostic slug used in error messages.
    tol
        Float comparison tolerance.
    replacements
        Mapping of ``{real_value: placeholder}`` applied to the actual document.

    Raises
    ------
    AssertionError
        If the bundles differ beyond tolerance after sanitisation.

    Notes
    -----
    If ``expected_path`` does not exist (a legacy regression without a committed bundle),
    the comparison is skipped silently and the function returns.
    """
    if not expected_path.exists():
        # Legacy regression data without this bundle file — skip silently
        return

    expected_bytes = expected_path.read_bytes()
    actual_bytes = actual_path.read_bytes()

    # Fast path: byte-identical means no difference.
    if expected_bytes == actual_bytes:
        return

    expected_obj = json.loads(expected_bytes.decode("utf-8"))
    actual_obj = json.loads(actual_bytes.decode("utf-8"))

    # Rewrite actual — both keys and leaf values — before comparison.
    actual_sanitised = _rewrite_keys_and_values(actual_obj, ordered_replacements(replacements))

    mismatches = compare_json_content(expected_obj, actual_sanitised, tol=tol)
    if mismatches:
        mismatch_detail = "\n  ".join(mismatches)
        msg = (
            f"Diagnostic {slug!r}: committed bundle {expected_path.name!r} "
            f"differs from regenerated output after sanitisation.\n"
            f"Mismatches ({len(mismatches)}):\n  {mismatch_detail}\n\n"
            f"Remediation: if the change is intentional, bump the diagnostic's "
            f"test_case_version and regenerate with `ref test-cases run --force-regen`."
        )
        raise AssertionError(msg)

build_native_snapshot(base_dir, relpaths)

Record a sha256 + size snapshot of each persisted native file.

Parameters:

Name	Type	Description	Default
base_dir	Path	The per-execution results directory the relpaths are resolved against.	required
relpaths	list[Path]	The persisted files (relative to base_dir), e.g. the return value of :func:~climate_ref_core.output_files.copy_execution_outputs.	required

Returns:

Type	Description
dict[str, NativeEntry]	Mapping of POSIX relpath -> :class:NativeEntry for every persisted file.

Source code in packages/climate-ref-core/src/climate_ref_core/regression/capture.py

def build_native_snapshot(base_dir: Path, relpaths: list[Path]) -> dict[str, NativeEntry]:
    """
    Record a sha256 + size snapshot of each persisted native file.

    Parameters
    ----------
    base_dir
        The per-execution results directory the relpaths are resolved against.
    relpaths
        The persisted files (relative to ``base_dir``), e.g. the return value of
        :func:`~climate_ref_core.output_files.copy_execution_outputs`.

    Returns
    -------
    :
        Mapping of POSIX relpath -> :class:`NativeEntry` for every persisted file.
    """
    entries: dict[str, NativeEntry] = {}
    for relpath in relpaths:
        path = base_dir / relpath
        entries[relpath.as_posix()] = NativeEntry(sha256=sha256_file(path), size=path.stat().st_size)
    return entries

build_native_store(config, *, writable)

Build a :class:NativeStore from a native-store config object.

Accepts any object that exposes url, cache_dir, s3_endpoint_url and bucket (satisfying :class:_NativeStoreConfigProtocol), so callers pass config.native_store rather than the full :class:~climate_ref.config.Config.

With writable=False the returned store is anonymous and credential-free, which suits the CI read and replay paths. With writable=True and a remote URL the S3 endpoint and bucket come from the config, and authentication is read from the environment (REF_NATIVE_STORE_ACCESS_KEY_ID / REF_NATIVE_STORE_SECRET_ACCESS_KEY, else REF_NATIVE_STORE_PROFILE, else boto3's default chain), so secrets never live in the persisted config. A local store is always readable and writable, so writable makes no difference to it.

Parameters:

Name	Type	Description	Default
config	_NativeStoreConfigProtocol	A config object providing url, cache_dir, s3_endpoint_url and bucket. Typically app_config.native_store.	required
writable	bool	Whether the store must be able to write.	required

Returns:

Type	Description
NativeStore	The configured store.

Raises:

Type	Description
ValueError	If the URL scheme is unrecognised, or a writable remote store is requested without an S3 endpoint / bucket configured.

Source code in packages/climate-ref-core/src/climate_ref_core/regression/store.py

def build_native_store(config: _NativeStoreConfigProtocol, *, writable: bool) -> NativeStore:
    """
    Build a :class:`NativeStore` from a native-store config object.

    Accepts any object that exposes ``url``, ``cache_dir``, ``s3_endpoint_url`` and ``bucket``
    (satisfying :class:`_NativeStoreConfigProtocol`), so callers pass ``config.native_store``
    rather than the full :class:`~climate_ref.config.Config`.

    With ``writable=False`` the returned store is anonymous and credential-free,
    which suits the CI read and replay paths.
    With ``writable=True`` and a remote URL the S3 endpoint and bucket come from the config,
    and authentication is read from the environment
    (``REF_NATIVE_STORE_ACCESS_KEY_ID`` / ``REF_NATIVE_STORE_SECRET_ACCESS_KEY``,
    else ``REF_NATIVE_STORE_PROFILE``, else boto3's default chain),
    so secrets never live in the persisted config.
    A local store is always readable and writable, so ``writable`` makes no difference to it.

    Parameters
    ----------
    config
        A config object providing ``url``, ``cache_dir``, ``s3_endpoint_url`` and ``bucket``.
        Typically ``app_config.native_store``.
    writable
        Whether the store must be able to write.

    Returns
    -------
    :
        The configured store.

    Raises
    ------
    ValueError
        If the URL scheme is unrecognised, or a writable remote store is requested
        without an S3 endpoint / bucket configured.
    """
    store = NativeStore(url=config.url, cache_dir=config.cache_dir)
    if not writable or store.root is not None:
        # A local store is already writable, and a read-only store needs no credentials.
        return store
    return NativeStore(
        url=config.url,
        cache_dir=config.cache_dir,
        write=S3WriteConfig(
            endpoint_url=config.s3_endpoint_url,
            bucket=config.bucket,
            access_key_id=os.environ.get("REF_NATIVE_STORE_ACCESS_KEY_ID", ""),
            secret_access_key=os.environ.get("REF_NATIVE_STORE_SECRET_ACCESS_KEY", ""),
            profile=os.environ.get("REF_NATIVE_STORE_PROFILE", ""),
        ),
    )

compare_json_content(expected, actual, *, tol, path='')

Recursively compare two parsed JSON values with float tolerance.

Rules: - Floats: compared with relative tolerance tol.rtol and absolute tolerance tol.atol. - Ints, strings, bools, None: exact equality. - Lists: element-by-element, same length required. - Dicts: key sets must match; values compared recursively.

Parameters:

Name	Type	Description	Default
expected	Any	The reference (committed) parsed JSON value.	required
actual	Any	The regenerated parsed JSON value.	required
tol	Tolerance	Float comparison tolerance.	required
path	str	Dot-/bracket-notation path prefix for error messages (empty at top level).	''

Returns:

Type	Description
list[str]	A list of human-readable mismatch descriptions. An empty list means the values are equivalent within tolerance.

Source code in packages/climate-ref-core/src/climate_ref_core/regression/compare.py

def compare_json_content(
    expected: Any,
    actual: Any,
    *,
    tol: Tolerance,
    path: str = "",
) -> list[str]:
    """
    Recursively compare two parsed JSON values with float tolerance.

    Rules:
    - **Floats**: compared with relative tolerance ``tol.rtol``
      and absolute tolerance ``tol.atol``.
    - **Ints, strings, bools, ``None``**: exact equality.
    - **Lists**: element-by-element, same length required.
    - **Dicts**: key sets must match; values compared recursively.

    Parameters
    ----------
    expected
        The reference (committed) parsed JSON value.
    actual
        The regenerated parsed JSON value.
    tol
        Float comparison tolerance.
    path
        Dot-/bracket-notation path prefix for error messages (empty at top level).

    Returns
    -------
    :
        A list of human-readable mismatch descriptions.
        An empty list means the values are equivalent within tolerance.
    """
    mismatches: list[str] = []
    _compare_recursive(expected, actual, tol=tol, path=path, out=mismatches)
    return mismatches

compute_committed_digests(regression_dir)

Compute sha256 digests of the committed regression JSON artefacts.

The digests are taken over the bytes exactly as they sit on disk (placeholder text included), so a CI recompute is deterministic. Only files that exist are included.

Parameters:

Name	Type	Description	Default
regression_dir	Path	The test case regression/ directory.	required

Returns:

Type	Description
dict[str, str]	Mapping of {relpath: sha256} for each present committed artefact.

Source code in packages/climate-ref-core/src/climate_ref_core/regression/manifest.py

def compute_committed_digests(regression_dir: Path) -> dict[str, str]:
    """
    Compute sha256 digests of the committed regression JSON artefacts.

    The digests are taken over the bytes exactly as they sit on disk (placeholder text included),
    so a CI recompute is deterministic. Only files that exist are included.

    Parameters
    ----------
    regression_dir
        The test case ``regression/`` directory.

    Returns
    -------
    :
        Mapping of ``{relpath: sha256}`` for each present committed artefact.
    """
    digests: dict[str, str] = {}
    for relpath in COMMITTED_BUNDLE_FILES:
        candidate = regression_dir / relpath
        if candidate.exists():
            digests[relpath] = sha256_file(candidate)
    return digests

decide_coupling(manifest, base_manifest, *, extraction_changed=False, committed_integrity_ok=True, catalog_integrity_ok=True, code_diagnostic_version=None)

Decide how CI should verify a single test case's regression baseline.

The decision is pure: it performs no I/O. All on-disk reality is summarised by its arguments, namely the two manifests, whether the diagnostic's extraction code changed, and whether the committed bundle on disk still matches the current manifest's digests.

The gate fails if any state cannot be positively verified. A deleted managed manifest, a committed bundle that drifted from its manifest, an input catalog that drifted from its manifest is a failure rather than a silent skip.

Changes to the native baseline are not failures. This is due to the workflow for minting requires credentials. This means fork contributors cannot author or edit native blobs. replay is therefore only selected when native blobs actually exist to replay. An absent or removed native baseline downgrades to skip, never fail.

See the module docstring for the meaning of each :class:Action.

Parameters:

Name	Type	Description	Default
manifest	Manifest | None	The current manifest.json for the test case, or None if the case has no manifest on this branch (never managed, or deleted in this change).	required
base_manifest	Manifest | None	The manifest.json as it exists on the base branch, or None if the manifest is newly added in this change (seeding).	required
extraction_changed	bool	Whether code that influences build_execution_result for this test case changed in this pull request. Only consulted when the committed bundle is unchanged and the version was not bumped.	False
committed_integrity_ok	bool	Whether the committed bundle on disk matches the current manifest's committed digests exactly (no edited, added, or removed committed file). The caller computes this against the working tree. False means the manifest no longer describes the bundle it is supposed to gate, which is a hard failure. Ignored when manifest is None (nothing to verify).	True
catalog_integrity_ok	bool	Whether the test case's input catalog.yaml still matches the current manifest's catalog_hash. The caller computes this against the working tree. False means the inputs changed without the baseline being regenerated, so the committed bundle no longer reflects its inputs, a hard failure. Always True when the manifest carries no catalog_hash	True
code_diagnostic_version	int | None	The diagnostic's current in-code Diagnostic.version, injected by the caller (the pure function does no I/O). When provided, a code version that exceeds manifest.diagnostic_version fails the case as a stale baseline that must be re-minted, and a code version below it fails because Diagnostic.version must not drop below the minted baseline's version. None (the default) skips the staleness comparison entirely, preserving legacy behaviour.	None

Returns:

Type	Description
GateDecision	The gate's decision, pairing an :class:Action with a reason.

Source code in packages/climate-ref-core/src/climate_ref_core/regression/gate.py

def decide_coupling(  # noqa: PLR0911, PLR0912, PLR0913
    manifest: Manifest | None,
    base_manifest: Manifest | None,
    *,
    extraction_changed: bool = False,
    committed_integrity_ok: bool = True,
    catalog_integrity_ok: bool = True,
    code_diagnostic_version: int | None = None,
) -> GateDecision:
    """
    Decide how CI should verify a single test case's regression baseline.

    The decision is pure: it performs no I/O.
    All on-disk reality is summarised by its arguments,
    namely the two manifests, whether the diagnostic's extraction code
    changed, and whether the committed bundle on disk still matches the current
    manifest's digests.

    The gate fails if any state cannot be positively verified.
    A deleted managed manifest, a committed bundle that drifted from its manifest,
    an input catalog that drifted from its manifest
    is a failure rather than a silent skip.

    Changes to the native baseline are not failures.
    This is due to the workflow for minting requires credentials.
    This means fork contributors cannot author or edit native blobs.
    ``replay`` is therefore only selected when native blobs actually exist to replay.
    An absent or removed native baseline downgrades to ``skip``, never ``fail``.

    See the module docstring for the meaning of each :class:`Action`.

    Parameters
    ----------
    manifest
        The current ``manifest.json`` for the test case,
        or ``None`` if the case has no manifest on this branch (never managed, or deleted in this change).
    base_manifest
        The ``manifest.json`` as it exists on the base branch,
        or ``None`` if the manifest is newly added in this change (seeding).
    extraction_changed
        Whether code that influences ``build_execution_result`` for this test case
        changed in this pull request.
        Only consulted when the committed bundle is unchanged and the version was not bumped.
    committed_integrity_ok
        Whether the committed bundle on disk matches the current manifest's ``committed`` digests exactly
        (no edited, added, or removed committed file).
        The caller computes this against the working tree.
        ``False`` means the manifest no longer describes the bundle it is supposed to gate,
        which is a hard failure.

        Ignored when ``manifest`` is ``None`` (nothing to verify).
    catalog_integrity_ok
        Whether the test case's input ``catalog.yaml`` still matches the current
        manifest's ``catalog_hash``.
        The caller computes this against the working tree.
        ``False`` means the inputs changed without the baseline being regenerated,
        so the committed bundle no longer reflects its inputs, a hard failure.

        Always ``True`` when the manifest carries no ``catalog_hash``
    code_diagnostic_version
        The diagnostic's current in-code ``Diagnostic.version``, injected by the caller
        (the pure function does no I/O). When provided, a code version that exceeds
        ``manifest.diagnostic_version`` fails the case as a stale baseline that must be
        re-minted, and a code version below it fails because ``Diagnostic.version`` must
        not drop below the minted baseline's version.
        ``None`` (the default) skips the staleness comparison entirely, preserving legacy behaviour.

    Returns
    -------
    :
        The gate's decision, pairing an :class:`Action` with a reason.
    """
    if manifest is None:
        # Distinguish a never-managed case from the deletion of a managed baseline.
        # Deleting manifest.json must not be a silent way to disable the gate.
        if base_manifest is not None:
            return GateDecision(
                Action.FAIL,
                "manifest.json is absent but exists on the base branch; a managed "
                "regression baseline cannot be removed without review",
            )
        return GateDecision(
            Action.SKIP,
            "no manifest.json; test case not under regression-baseline management",
        )

    # The manifest must faithfully describe the committed bundle it gates.
    # If the bundle on disk drifted from the manifest digests
    # (an edit/add/remove without regenerating the manifest), the metadata comparisons below are meaningless.
    if not committed_integrity_ok:
        return GateDecision(
            Action.FAIL,
            "committed bundle on disk does not match manifest.json digests; "
            "regenerate the manifest with `ref test-cases run` after changing the bundle",
        )

    # The input catalog must still describe the baseline it produced.
    if not catalog_integrity_ok:
        return GateDecision(
            Action.FAIL,
            "input catalog.yaml does not match manifest.json catalog_hash; "
            "regenerate the baseline with `ref test-cases run` after changing the inputs",
        )

    # Runs before seeding so a newly added manifest whose author bumped the code
    # without re-minting still fails.
    # Equality must fall through; only the > and < arms return.
    if manifest is not None and code_diagnostic_version is not None:
        if code_diagnostic_version > manifest.diagnostic_version:
            return GateDecision(
                Action.FAIL,
                f"diagnostic Diagnostic.version is {code_diagnostic_version} but the baseline "
                f"manifest records diagnostic_version {manifest.diagnostic_version}; the "
                "diagnostic's results changed; re-mint this baseline (`ref test-cases mint`) "
                "so the committed bundle reflects the new version",
            )
        if code_diagnostic_version < manifest.diagnostic_version:
            return GateDecision(
                Action.FAIL,
                f"diagnostic Diagnostic.version ({code_diagnostic_version}) is below the "
                f"baseline's diagnostic_version ({manifest.diagnostic_version}); "
                "Diagnostic.version must not be below the minted baseline's version, "
                "re-mint after changing it",
            )

    if base_manifest is None:
        # Seeding a newly added manifest.
        # Replay only verifies something when native blobs exist
        if manifest.native:
            return GateDecision(
                Action.REPLAY,
                "manifest newly added (seeding); replaying native baseline against committed bundle",
            )
        return GateDecision(
            Action.SKIP,
            "manifest newly added (seeding) with no native baseline; "
            "the committed bundle is the only signal and is reviewed in the diff",
        )

    if manifest.test_case_version < base_manifest.test_case_version:
        return GateDecision(
            Action.FAIL,
            f"test_case_version decreased ({base_manifest.test_case_version} -> "
            f"{manifest.test_case_version}); version must be monotonic",
        )

    if manifest.diagnostic_version < base_manifest.diagnostic_version:
        return GateDecision(
            Action.FAIL,
            f"diagnostic_version decreased ({base_manifest.diagnostic_version} -> "
            f"{manifest.diagnostic_version}), but the version must be monotonic",
        )

    version_bumped = manifest.test_case_version > base_manifest.test_case_version
    committed_changed = manifest.committed != base_manifest.committed
    native_changed = manifest.native != base_manifest.native

    if version_bumped:
        version_change = f"{base_manifest.test_case_version} -> {manifest.test_case_version}"
        if manifest.native:
            # A native baseline ships with the bump, so the replay can prove the new committed bundle
            # actually reproduces from those blobs.
            return GateDecision(
                Action.REPLAY,
                f"test_case_version bumped ({version_change}) with a native baseline present; "
                "replaying to confirm the native baseline reproduces the new committed bundle",
            )
        return GateDecision(
            Action.SKIP,
            f"test_case_version bumped ({version_change}) with no native baseline to replay, "
            "so the committed bundle changed with nothing to verify it against and the diff "
            "review is the only signal",
        )

    if committed_changed:
        return GateDecision(
            Action.FAIL,
            "committed bundle changed without a test_case_version bump; "
            "bump test_case_version to authorise the new baseline",
        )

    if native_changed or extraction_changed:
        trigger = "native baseline changed" if native_changed else "extraction code changed"
        if manifest.native:
            return GateDecision(
                Action.REPLAY,
                f"{trigger} with committed bundle unchanged, "
                "replaying to confirm the native baseline reproduces the committed bundle",
            )
        # Native changed to empty (de-mint) or was already empty: nothing to replay against.
        return GateDecision(
            Action.SKIP,
            f"{trigger} but no native baseline exists to replay, "
            "so the unchanged committed bundle remains the only signal",
        )

    return GateDecision(
        Action.SKIP,
        "no committed-bundle, native, version, or extraction-code change; nothing to verify",
    )

materialise_native(native, store, dest)

Materialise a native snapshot from a store into a destination directory.

For each (relpath, entry) the blob is fetched from store (keyed by its sha256 digest) to dest / relpath, creating parent directories as needed.

Parameters:

Name	Type	Description	Default
native	dict[str, NativeEntry]	Mapping of relpath -> :class:NativeEntry (from a manifest).	required
store	NativeStore	A content-addressed :class:~climate_ref_core.regression.store.NativeStore.	required
dest	Path	The destination directory the snapshot is materialised into.	required

Source code in packages/climate-ref-core/src/climate_ref_core/regression/capture.py

def materialise_native(native: dict[str, NativeEntry], store: NativeStore, dest: Path) -> None:
    """
    Materialise a native snapshot from a store into a destination directory.

    For each ``(relpath, entry)`` the blob is fetched from ``store`` (keyed by its
    sha256 digest) to ``dest / relpath``, creating parent directories as needed.

    Parameters
    ----------
    native
        Mapping of relpath -> :class:`NativeEntry` (from a manifest).
    store
        A content-addressed :class:`~climate_ref_core.regression.store.NativeStore`.
    dest
        The destination directory the snapshot is materialised into.
    """
    for relpath, entry in native.items():
        # Defend against path traversal: a hand-edited or hostile manifest could
        # carry an absolute path or one with '..' components that escapes dest.
        target = safe_path(relpath, dest, label="native path")
        target.parent.mkdir(parents=True, exist_ok=True)
        store.fetch(entry.sha256, target)

paths_under(changed_files, roots)

Return whether any changed file lies within one of the given directory roots.

A small helper for deriving the extraction_changed signal from a pull request's changed-file list. Paths are compared textually as POSIX-style, repo-relative strings, so callers must normalise both changed_files and roots to the same convention (e.g. git diff --name-only output and a package source directory relative to the repo root).

Parameters:

Name	Type	Description	Default
changed_files	Iterable[str]	Repo-relative paths changed in the pull request.	required
roots	Iterable[str]	Repo-relative directory prefixes to test against. A trailing slash is optional; an empty root never matches.	required

Returns:

Type	Description
bool	True if any changed file equals or sits beneath any root.

Source code in packages/climate-ref-core/src/climate_ref_core/regression/gate.py

def paths_under(changed_files: Iterable[str], roots: Iterable[str]) -> bool:
    """
    Return whether any changed file lies within one of the given directory roots.

    A small helper for deriving the ``extraction_changed`` signal from a pull request's changed-file list.
    Paths are compared textually as POSIX-style, repo-relative strings,
    so callers must normalise both ``changed_files`` and ``roots`` to the same convention
    (e.g. ``git diff --name-only`` output and a package source directory relative to the repo root).

    Parameters
    ----------
    changed_files
        Repo-relative paths changed in the pull request.
    roots
        Repo-relative directory prefixes to test against. A trailing slash is
        optional; an empty root never matches.

    Returns
    -------
    :
        ``True`` if any changed file equals or sits beneath any root.
    """
    normalised_roots = [root.rstrip("/") for root in roots if root.rstrip("/")]
    for changed in changed_files:
        for root in normalised_roots:
            if changed == root or changed.startswith(root + "/"):
                return True
    return False

sha256_bytes(data)

Compute the sha256 digest of an in-memory byte string.

Parameters:

Name	Type	Description	Default
data	bytes	The bytes to hash.	required

Returns:

Type	Description
str	The hex-encoded sha256 digest.

Source code in packages/climate-ref-core/src/climate_ref_core/regression/manifest.py

def sha256_bytes(data: bytes) -> str:
    """
    Compute the sha256 digest of an in-memory byte string.

    Parameters
    ----------
    data
        The bytes to hash.

    Returns
    -------
    :
        The hex-encoded sha256 digest.
    """
    return hashlib.sha256(data).hexdigest()

sha256_file(path)

Compute the sha256 digest of a file.

Reuses :func:pooch.hashes.file_hash so the digest agrees with pooch elsewhere.

Parameters:

Name	Type	Description	Default
path	Path	Path to the file to hash.	required

Returns:

Type	Description
str	The hex-encoded sha256 digest.

Source code in packages/climate-ref-core/src/climate_ref_core/regression/manifest.py

def sha256_file(path: Path) -> str:
    """
    Compute the sha256 digest of a file.

    Reuses :func:`pooch.hashes.file_hash` so the digest agrees with pooch elsewhere.

    Parameters
    ----------
    path
        Path to the file to hash.

    Returns
    -------
    :
        The hex-encoded sha256 digest.
    """
    return pooch.hashes.file_hash(str(path), alg="sha256")

verify_committed_integrity(manifest, regression_dir)

Check that the committed regression artefacts match the manifest digests.

Used by the CI integrity check. An empty return value means the bundle is intact.

Parameters:

Name	Type	Description	Default
manifest	Manifest	The manifest holding the expected committed digests.	required
regression_dir	Path	The test case regression/ directory to verify against.	required

Returns:

Type	Description
list[str]	A list of human-readable mismatch descriptions; empty when everything matches.

Source code in packages/climate-ref-core/src/climate_ref_core/regression/manifest.py

def verify_committed_integrity(manifest: Manifest, regression_dir: Path) -> list[str]:
    """
    Check that the committed regression artefacts match the manifest digests.

    Used by the CI integrity check. An empty return value means the bundle is intact.

    Parameters
    ----------
    manifest
        The manifest holding the expected committed digests.
    regression_dir
        The test case ``regression/`` directory to verify against.

    Returns
    -------
    :
        A list of human-readable mismatch descriptions; empty when everything matches.
    """
    mismatches: list[str] = []
    for relpath, expected in manifest.committed.items():
        candidate = regression_dir / relpath
        if not candidate.exists():
            mismatches.append(
                f"{relpath}: missing on disk — expected at {candidate} (manifest sha256 {expected})"
            )
            continue
        actual = sha256_file(candidate)
        if actual != expected:
            mismatches.append(
                f"{relpath}: content differs from manifest — {candidate} "
                f"(manifest sha256 {expected}, on-disk sha256 {actual})"
            )
    return mismatches

write_committed_bundle(source_dir, regression_dir, *, placeholders)

Write the sanitised committed CMEC bundle into regression_dir.

Copies each committed artefact present in source_dir into regression_dir, rewrites absolute paths to portable placeholders in place (:meth:~climate_ref_core.output_files.PlaceholderMap.sanitise), then canonicalises every committed JSON file -- rounding floats and redacting host/user CMEC provenance into a deterministic on-disk form (:func:_canonicalise_committed_bundle). When a committed artefact is absent from source_dir, any stale copy left in regression_dir from a previous capture is removed so it is not re-digested.

Parameters:

Name	Type	Description	Default
source_dir	Path	Directory holding the freshly persisted CMEC artefacts (the per-execution results directory).	required
regression_dir	Path	The destination regression/ directory (created if needed).	required
placeholders	PlaceholderMap	The placeholder map for this execution, already bound to the output directory via :meth:~climate_ref_core.output_files.PlaceholderMap.with_output. Its absolute paths are rewritten to portable <TOKEN> placeholders in the copied bundle.	required

Returns:

Type	Description
dict[str, str]	The committed digests {filename: sha256} of the bytes just written, suitable for :attr:Manifest.committed.

Raises:

Type	Description
ValueError	If placeholders is not bound to an output directory. An unbound map would leave execution-specific output paths in the committed bundle and digest those machine-specific bytes.

Source code in packages/climate-ref-core/src/climate_ref_core/regression/capture.py

def write_committed_bundle(
    source_dir: Path,
    regression_dir: Path,
    *,
    placeholders: PlaceholderMap,
) -> dict[str, str]:
    """
    Write the sanitised committed CMEC bundle into ``regression_dir``.

    Copies each committed artefact present in ``source_dir`` into ``regression_dir``,
    rewrites absolute paths to portable placeholders in place
    (:meth:`~climate_ref_core.output_files.PlaceholderMap.sanitise`),
    then canonicalises every committed JSON file -- rounding floats and redacting host/user CMEC
    provenance into a deterministic on-disk form (:func:`_canonicalise_committed_bundle`).
    When a committed artefact is absent from ``source_dir``,
    any stale copy left in ``regression_dir`` from a previous capture is removed so it is not re-digested.

    Parameters
    ----------
    source_dir
        Directory holding the freshly persisted CMEC artefacts (the per-execution
        results directory).
    regression_dir
        The destination ``regression/`` directory (created if needed).
    placeholders
        The placeholder map for this execution, already bound to the output directory via
        :meth:`~climate_ref_core.output_files.PlaceholderMap.with_output`. Its absolute paths are
        rewritten to portable ``<TOKEN>`` placeholders in the copied bundle.

    Returns
    -------
    :
        The committed digests ``{filename: sha256}`` of the bytes just written,
        suitable for :attr:`Manifest.committed`.

    Raises
    ------
    ValueError
        If ``placeholders`` is not bound to an output directory.
        An unbound map would leave execution-specific output paths in the committed bundle
        and digest those machine-specific bytes.
    """
    if not placeholders.is_output_bound:
        raise ValueError(
            "placeholders must be bound to an output directory via with_output() "
            "before writing a committed bundle"
        )

    regression_dir.mkdir(parents=True, exist_ok=True)

    for filename in COMMITTED_BUNDLE_FILES:
        source = source_dir / filename
        dest = regression_dir / filename
        if source.exists():
            shutil.copy(source, dest)
        else:
            # Drop a stale copy from a previous capture so it is not re-digested.
            dest.unlink(missing_ok=True)

    placeholders.sanitise(regression_dir)
    # Canonicalise every committed file (round floats, redact provenance, re-dump deterministically)
    # before digesting, so the recorded digests are over the stable, portable bytes. Placeholder
    # substitution only rewrites path strings, so its order relative to canonicalisation is immaterial.
    _canonicalise_committed_bundle(regression_dir)
    return compute_committed_digests(regression_dir)

sub-packages

Sub-package	Description
_quantise	Float quantisation for committed regression bundles.
capture	Capture of regression baselines from a diagnostic execution.
compare	Content comparison utilities for regression testing.
gate	CI coupling gate for test case regression bundles.
manifest	Manifest model and digest utilities for test case regression bundles.
store	Data store for native bundles.