Source code for aeat.domain.calculations.registry._loader_cache

"""Registry loader cache predicates.

This module centralizes the small policy decisions that keep registry loading
fast without hiding live TOML edits. Bundled registry roots receive a short
fingerprint TTL, mutable authoring roots keep the stricter window, and under
pytest the cross-process disk pickle is shared only for the immutable
bundled root -- a mutable/synthetic root always keeps it disabled so xdist
workers cannot share a stale compiled registry from a tree the run itself
can edit.

See Also:
    :mod:`~domain.calculations.registry._loader`
        Registry TOML loader that consumes these TTL and disk-cache predicates.
    :func:`~core.resources.bundled_path`
        Resource boundary used to identify the package-bundled registry root.
    :func:`~domain.calculations.registry.tests.test_loader_cache_isolation.test_registry_disk_cache_disabled_under_pytest`
        Real-behavior gate for the pytest disk-cache refusal path.
    :func:`~domain.calculations.registry.tests.test_loader_cache_isolation.test_is_bundled_registry_root_rejects_a_mutable_authoring_tree`
        Coverage for bundled-root versus mutable-authoring-tree separation.
    :func:`~domain.calculations.registry.tests.test_loader_cache_isolation.test_bundled_tree_fingerprint_cache_survives_past_the_mutable_tree_ttl`
        Coverage for the longer bundled-root fingerprint TTL window.
    :func:`~conftest._isolate_registry_caches`
        Session fixture that clears registry caches around pytest runs.
    Governing vault records
        ``2026-05-20-registry-authority-flow-adr`` and
        ``2026-06-02-registry-loader-boundary-audit`` govern the authority
        boundary and loader-extraction cache invalidation behavior.
"""

from __future__ import annotations

import tempfile
from functools import lru_cache
from pathlib import Path

from ....core.config import load_settings
from ....core.resources import bundled_path

REGISTRY_DISK_CACHE_DIR_ENV_VAR = "AEAT_REGISTRY_DISK_CACHE_DIR"
"""Environment variable backing :attr:`~core.config.Settings.aeat_registry_disk_cache_dir`."""

# The bundled tree gets a longer fingerprint TTL than a mutable authoring
# tree, but NOT a process-lifetime one: under an editable install (the
# routine dev/worktree mode) "bundled" resolves to the literal in-tree
# ``src/aeat/_data/registry/aeat`` source directory, which concurrent peer
# agents in this shared worktree edit live throughout a session. A TTL that
# never re-checks would silently serve stale registry TOML to a long-running
# process (an MCP server, a REPL, a background watch loop) after such an
# edit lands. 10 seconds is long enough to fold the several fingerprint
# recomputations one calculate call triggers (authority + snapshot + any
# nested revision lookups, all milliseconds apart) into a single directory
# walk, while still re-scanning promptly enough that a concurrent registry
# edit is picked up well within one operator interaction. A genuinely
# read-only installed (non-editable) wheel benefits identically: nothing
# ever rewrites it, so the periodic re-walk merely repeats the same answer.
BUNDLED_REGISTRY_FINGERPRINT_TTL_SECONDS = 10.0
MUTABLE_REGISTRY_FINGERPRINT_TTL_SECONDS = 1.0


@lru_cache(maxsize=1)
def _bundled_registry_root() -> Path:
    """Return the resolved package-bundled registry root, computed once per process."""
    return bundled_path("registry", "aeat").resolve()




def is_bundled_registry_root(resolved: Path) -> bool:
    """Whether ``resolved`` is the package-bundled registry tree.

    The bundled tree is shipped inside the installed wheel (or, under an
    editable install, force-included from the in-tree ``registry/aeat``
    directory) rather than passed explicitly as a mutable authoring tree
    (e.g. a test's ``tmp_path`` fixture building a synthetic registry).
    Comparing the resolved path against the bundled root lets the
    fingerprint cache apply :data:`BUNDLED_REGISTRY_FINGERPRINT_TTL_SECONDS`
    to the bundled tree alone without weakening invalidation for any
    mutable tree, which always keeps the strict
    :data:`MUTABLE_REGISTRY_FINGERPRINT_TTL_SECONDS` window.
    """
    try:
        return resolved == _bundled_registry_root()
    except (ImportError, OSError, ValueError):
        # A resources boundary failure (e.g. no bundled data under an unusual
        # install) must never be mistaken for "this is the bundled tree";
        # fail closed to the strict mutable-tree TTL.
        return False





def registry_disk_cache_enabled(*, is_bundled: bool = False) -> bool:
    """Whether the cross-process ``/tmp`` registry pickle is read/written.

    Production (no pytest markers present at all) always keeps the disk
    cache: it loads the registry once at startup with no concurrent edits.

    Under pytest, including collection before ``PYTEST_CURRENT_TEST`` is
    set, the cache is enabled ONLY for ``is_bundled=True`` -- the
    package-bundled, read-only registry tree (:func:`is_bundled_registry_root`).
    That tree is never mutated during a test run, so every pytest-xdist
    worker and every subprocess-spawning test may safely share ONE compiled
    pickle keyed by a content fingerprint of that tree, collapsing what would
    otherwise be an independent multi-second cold compile per worker/subprocess
    into a single shared compile the rest read.

    A mutable or synthetic root (e.g. a test's ``tmp_path`` registry, or any
    path that is not the resolved bundled root) always keeps the cache
    disabled under pytest -- this is the #44 isolation fix: such a root CAN be
    edited mid-run by the very test that built it, and the pickle is keyed by
    file mtime, so sharing it across workers could serve a stale or
    transiently-inconsistent compiled registry (the M303-2009 flake #44
    diagnosed). Only the always-immutable bundled tree is exempt from that
    race.
    """
    import os
    import sys

    under_pytest = (
        "pytest" in sys.modules
        or "PYTEST_CURRENT_TEST" in os.environ
        or "PYTEST_XDIST_WORKER" in os.environ
        or "PYTEST_VERSION" in os.environ
    )
    if under_pytest:
        return is_bundled
    return True





def registry_disk_cache_dir() -> Path:
    """Return the directory the cross-process registry disk pickle lives in.

    Reads :attr:`~core.config.Settings.aeat_registry_disk_cache_dir` (the
    ``AEAT_REGISTRY_DISK_CACHE_DIR`` env var) before falling back to
    ``tempfile.gettempdir()``, so a test can redirect the disk-cache pickle to
    a test-owned directory. Production and the ordinary bundled-root sharing
    path never set this field and always use the real OS temp directory; only
    a test that needs to assert EXCLUSIVE state on the pickle (e.g. "exactly
    one file exists", "the mtime is unchanged") -- which the real OS temp
    directory cannot guarantee once sibling pytest-xdist workers are also
    touching the shared bundled-root pickle -- sets this var to isolate its
    own assertions from that sibling traffic, while still exercising the real
    filesystem and the real pickle read/write path (no mock of the loader's
    own behavior). It rides the env var (rather than a plain monkeypatched
    function) because it also needs to propagate to a subprocess a test
    spawns via ``env=``, so a cross-process sharing proof can isolate BOTH
    ends of the process pair onto the same test-owned directory.
    """
    override = load_settings().aeat_registry_disk_cache_dir
    if override is not None:
        return override
    return Path(tempfile.gettempdir())