cache.py

"""Cache infrastructure primitives and graph-level helpers for TNFR.

This module consolidates structural cache helpers that previously lived in
legacy helper modules and are now exposed under :mod:`tnfr.utils`. The
functions exposed here are responsible for maintaining deterministic node
digests, scoped graph caches guarded by locks, and version counters that keep
edge artifacts in sync with ΔNFR driven updates.
"""

from __future__ import annotations

import asyncio
import hashlib
import logging
import pickle
import sys
import threading
import time
import weakref
from collections import defaultdict
from collections.abc import (
    Callable,
    Hashable,
    Iterable,
    Iterator,
    Mapping,
    MutableMapping,
)
from contextlib import contextmanager
from contextvars import ContextVar
from dataclasses import dataclass, field
from functools import lru_cache, wraps
from time import perf_counter
from typing import Any, Generic, TypeVar, cast

import networkx as nx

from ..errors import TNFRSecurityError, TNFRSecurityWarning, TNFRValueError
from ..locking import get_lock
from ..security.crypto import create_hmac_signer, create_hmac_validator
from ..types import CacheLevel
from ..types import CacheStats as CacheStatistics
from ..types import GraphLike, NodeId, TimingContext, TNFRGraph
from .cache_layers import (
    CacheLayer,
    MappingCacheLayer,
    RedisCacheLayer,
    ShelveCacheLayer,
    create_secure_redis_layer,
    create_secure_shelve_layer,
)
from .graph import get_graph, mark_dnfr_prep_dirty
from .unified_cache import UnifiedLRUCache

K = TypeVar("K", bound=Hashable)
V = TypeVar("V")
T = TypeVar("T")

SecurityError = TNFRSecurityError
SecurityWarning = TNFRSecurityWarning

__all__ = (
    "CacheLayer",
    "CacheManager",
    "CacheCapacityConfig",
    "CacheStatistics",
    "InstrumentedLRUCache",
    "ManagedLRUCache",
    "MappingCacheLayer",
    "RedisCacheLayer",
    "ShelveCacheLayer",
    "SecurityError",
    "SecurityWarning",
    "create_hmac_signer",
    "create_hmac_validator",
    "create_secure_shelve_layer",
    "create_secure_redis_layer",
    "prune_lock_mapping",
    "EdgeCacheManager",
    "NODE_SET_CHECKSUM_KEY",
    "cached_node_list",
    "cached_nodes_and_A",
    "clear_node_repr_cache",
    "edge_version_cache",
    "edge_version_update",
    "ensure_node_index_map",
    "ensure_node_offset_map",
    "stable_node_offsets",
    "get_graph_version",
    "increment_edge_version",
    "increment_graph_version",
    "node_set_checksum",
    "stable_json",
    "configure_graph_cache_limits",
    "DNFR_PREP_STATE_KEY",
    "GRAPH_RUNTIME_CACHE_KEYS",
    "DnfrPrepState",
    "build_cache_manager",
    "configure_global_cache_layers",
    "reset_global_cache_manager",
    "_GRAPH_CACHE_LAYERS_KEY",
    "_SeedHashCache",
    "ScopedCounterCache",
    "DnfrCache",
    "new_dnfr_cache",
    # Hierarchical cache classes (moved from caching/)
    "CacheLevel",
    "CacheEntry",
    "TNFRHierarchicalCache",
    # Cache decorators (moved from caching/decorators.py)
    "cache_tnfr_computation",
    "invalidate_function_cache",
    "get_global_cache",
    "set_global_cache",
    "reset_global_cache",
    # Invalidation tracking (moved from caching/invalidation.py)
    "GraphChangeTracker",
    "track_node_property_update",
    # Persistence (moved from caching/persistence.py)
    "PersistentTNFRCache",
)

# Environment variable to control security warnings for pickle deserialization
_TNFR_ALLOW_UNSIGNED_PICKLE = "TNFR_ALLOW_UNSIGNED_PICKLE"


@dataclass(frozen=True)
class CacheCapacityConfig:
    """Configuration snapshot for cache capacity policies."""

    default_capacity: int | None
    overrides: dict[str, int | None]


@dataclass
class DnfrCache:
    idx: dict[Any, int]
    theta: list[float]
    epi: list[float]
    vf: list[float]
    cos_theta: list[float]
    sin_theta: list[float]
    neighbor_x: list[float]
    neighbor_y: list[float]
    neighbor_epi_sum: list[float]
    neighbor_vf_sum: list[float]
    neighbor_count: list[float]
    neighbor_deg_sum: list[float] | None
    th_bar: list[float] | None = None
    epi_bar: list[float] | None = None
    vf_bar: list[float] | None = None
    deg_bar: list[float] | None = None
    degs: dict[Any, float] | None = None
    deg_list: list[float] | None = None
    theta_np: Any | None = None
    epi_np: Any | None = None
    vf_np: Any | None = None
    cos_theta_np: Any | None = None
    sin_theta_np: Any | None = None
    deg_array: Any | None = None
    edge_src: Any | None = None
    edge_dst: Any | None = None
    checksum: Any | None = None
    neighbor_x_np: Any | None = None
    neighbor_y_np: Any | None = None
    neighbor_epi_sum_np: Any | None = None
    neighbor_vf_sum_np: Any | None = None
    neighbor_count_np: Any | None = None
    neighbor_deg_sum_np: Any | None = None
    th_bar_np: Any | None = None
    epi_bar_np: Any | None = None
    vf_bar_np: Any | None = None
    deg_bar_np: Any | None = None
    grad_phase_np: Any | None = None
    grad_epi_np: Any | None = None
    grad_vf_np: Any | None = None
    grad_topo_np: Any | None = None
    grad_total_np: Any | None = None
    dense_components_np: Any | None = None
    dense_accum_np: Any | None = None
    dense_degree_np: Any | None = None
    neighbor_accum_np: Any | None = None
    neighbor_inv_count_np: Any | None = None
    neighbor_cos_avg_np: Any | None = None
    neighbor_sin_avg_np: Any | None = None
    neighbor_mean_tmp_np: Any | None = None
    neighbor_mean_length_np: Any | None = None
    edge_signature: Any | None = None
    neighbor_accum_signature: Any | None = None
    neighbor_edge_values_np: Any | None = None


def new_dnfr_cache() -> DnfrCache:
    """Return an empty :class:`DnfrCache` prepared for ΔNFR orchestration."""

    return DnfrCache(
        idx={},
        theta=[],
        epi=[],
        vf=[],
        cos_theta=[],
        sin_theta=[],
        neighbor_x=[],
        neighbor_y=[],
        neighbor_epi_sum=[],
        neighbor_vf_sum=[],
        neighbor_count=[],
        neighbor_deg_sum=[],
    )


@dataclass
class _CacheMetrics:
    hits: int = 0
    misses: int = 0
    evictions: int = 0
    total_time: float = 0.0
    timings: int = 0
    lock: threading.Lock = field(default_factory=threading.Lock, repr=False)

    def snapshot(self) -> CacheStatistics:
        return CacheStatistics(
            hits=self.hits,
            misses=self.misses,
            evictions=self.evictions,
            total_time=self.total_time,
            timings=self.timings,
        )


@dataclass
class _CacheEntry:
    factory: Callable[[], Any]
    lock: threading.Lock
    reset: Callable[[Any], Any] | None = None
    encoder: Callable[[Any], Any] | None = None
    decoder: Callable[[Any], Any] | None = None


class CacheManager:
    """Coordinate named caches guarded by per-entry locks."""

    _MISSING = object()

    def __init__(
        self,
        storage: MutableMapping[str, Any] | None = None,
        *,
        default_capacity: int | None = None,
        overrides: Mapping[str, int | None] | None = None,
        layers: Iterable[CacheLayer] | None = None,
    ) -> None:
        mapping_layer = MappingCacheLayer(storage)
        extra_layers: tuple[CacheLayer, ...]
        if layers is None:
            extra_layers = ()
        else:
            extra_layers = tuple(layers)
            for layer in extra_layers:
                if not isinstance(
                    layer, CacheLayer
                ):  # pragma: no cover - defensive typing
                    raise TypeError(f"unsupported cache layer type: {type(layer)!r}")
        self._layers: tuple[CacheLayer, ...] = (mapping_layer, *extra_layers)
        self._storage_layer = mapping_layer
        self._storage: MutableMapping[str, Any] = mapping_layer.storage
        self._entries: dict[str, _CacheEntry] = {}
        self._registry_lock = threading.RLock()
        self._default_capacity = self._normalise_capacity(default_capacity)
        self._capacity_overrides: dict[str, int | None] = {}
        self._metrics: dict[str, _CacheMetrics] = {}
        self._metrics_publishers: list[Callable[[str, CacheStatistics], None]] = []
        if overrides:
            self.configure(overrides=overrides)

    @staticmethod
    def _normalise_capacity(value: int | None) -> int | None:
        if value is None:
            return None
        size = int(value)
        if size < 0:
            raise TNFRValueError(
                "capacity must be non-negative or None",
                context={"capacity": value},
            )
        return size

    def register(
        self,
        name: str,
        factory: Callable[[], Any],
        *,
        lock_factory: Callable[[], threading.Lock | threading.RLock] | None = None,
        reset: Callable[[Any], Any] | None = None,
        create: bool = True,
        encoder: Callable[[Any], Any] | None = None,
        decoder: Callable[[Any], Any] | None = None,
    ) -> None:
        """Register `
        ame`` with ``factory`` and optional lifecycle hooks."""

        if lock_factory is None:
            lock_factory = threading.RLock
        with self._registry_lock:
            entry = self._entries.get(name)
            if entry is None:
                entry = _CacheEntry(
                    factory=factory,
                    lock=lock_factory(),
                    reset=reset,
                    encoder=encoder,
                    decoder=decoder,
                )
                self._entries[name] = entry
            else:
                # Update hooks when re-registering the same cache name.
                entry.factory = factory
                entry.reset = reset
                entry.encoder = encoder
                entry.decoder = decoder
            self._ensure_metrics(name)
        if create:
            self.get(name)

    def configure(
        self,
        *,
        default_capacity: int | None | object = _MISSING,
        overrides: Mapping[str, int | None] | None = None,
        replace_overrides: bool = False,
    ) -> None:
        """Update the cache capacity policy shared by registered entries."""

        with self._registry_lock:
            if default_capacity is not self._MISSING:
                self._default_capacity = self._normalise_capacity(
                    default_capacity if default_capacity is not None else None
                )
            if overrides is not None:
                if replace_overrides:
                    self._capacity_overrides.clear()
                for key, value in overrides.items():
                    self._capacity_overrides[key] = self._normalise_capacity(value)

    def configure_from_mapping(self, config: Mapping[str, Any]) -> None:
        """Load configuration produced by :meth:`export_config`."""

        default = config.get("default_capacity", self._MISSING)
        overrides = config.get("overrides")
        overrides_mapping: Mapping[str, int | None] | None
        overrides_mapping = overrides if isinstance(overrides, Mapping) else None
        self.configure(default_capacity=default, overrides=overrides_mapping)

    def export_config(self) -> CacheCapacityConfig:
        """Return a copy of the current capacity configuration."""

        with self._registry_lock:
            return CacheCapacityConfig(
                default_capacity=self._default_capacity,
                overrides=dict(self._capacity_overrides),
            )

    def get_capacity(
        self,
        name: str,
        *,
        requested: int | None = None,
        fallback: int | None = None,
        use_default: bool = True,
    ) -> int | None:
        """Return capacity for `
        ame`` considering overrides and defaults."""

        with self._registry_lock:
            override = self._capacity_overrides.get(name, self._MISSING)
            default = self._default_capacity
        if override is not self._MISSING:
            return override
        values: tuple[int | None, ...]
        if use_default:
            values = (requested, default, fallback)
        else:
            values = (requested, fallback)
        for value in values:
            if value is self._MISSING:
                continue
            normalised = self._normalise_capacity(value)
            if normalised is not None:
                return normalised
        return None

    def has_override(self, name: str) -> bool:
        """Return ``True`` if `
        ame`` has an explicit capacity override."""

        with self._registry_lock:
            return name in self._capacity_overrides

    def get_lock(self, name: str) -> threading.Lock | threading.RLock:
        """Return the lock guarding cache `
        ame`` for external coordination."""

        entry = self._entries.get(name)
        if entry is None:
            raise KeyError(name)
        return entry.lock

    def names(self) -> Iterator[str]:
        """Iterate over registered cache names."""

        with self._registry_lock:
            return iter(tuple(self._entries))

    def get(self, name: str, *, create: bool = True) -> Any:
        """Return cache `
        ame`` creating it on demand when ``create`` is true."""

        entry = self._entries.get(name)
        if entry is None:
            raise KeyError(name)
        with entry.lock:
            value = self._load_from_layers(name, entry)
            if create and value is None:
                value = entry.factory()
                self._persist_layers(name, entry, value)
            return value

    def peek(self, name: str) -> Any:
        """Return cache `
        ame`` without creating a missing entry."""

        entry = self._entries.get(name)
        if entry is None:
            raise KeyError(name)
        with entry.lock:
            return self._load_from_layers(name, entry)

    def store(self, name: str, value: Any) -> None:
        """Replace the stored value for cache `
        ame`` with ``value``."""

        entry = self._entries.get(name)
        if entry is None:
            raise KeyError(name)
        with entry.lock:
            self._persist_layers(name, entry, value)

    def update(
        self,
        name: str,
        updater: Callable[[Any], Any],
        *,
        create: bool = True,
    ) -> Any:
        """Apply ``updater`` to cache `
        ame`` storing the resulting value."""

        entry = self._entries.get(name)
        if entry is None:
            raise KeyError(name)
        with entry.lock:
            current = self._load_from_layers(name, entry)
            if create and current is None:
                current = entry.factory()
            new_value = updater(current)
            self._persist_layers(name, entry, new_value)
            return new_value

    def clear(self, name: str | None = None) -> None:
        """Reset caches either selectively or for every registered name."""

        if name is not None:
            names = (name,)
        else:
            with self._registry_lock:
                names = tuple(self._entries)
        for cache_name in names:
            entry = self._entries.get(cache_name)
            if entry is None:
                continue
            with entry.lock:
                current = self._load_from_layers(cache_name, entry)
                new_value = None
                if entry.reset is not None:
                    try:
                        new_value = entry.reset(current)
                    except Exception:  # pragma: no cover - defensive logging
                        _logger.exception("cache reset failed for %s", cache_name)
                if new_value is None:
                    try:
                        new_value = entry.factory()
                    except Exception:
                        self._delete_from_layers(cache_name)
                        continue
                self._persist_layers(cache_name, entry, new_value)

    # ------------------------------------------------------------------
    # Layer orchestration helpers

    def _encode_value(self, entry: _CacheEntry, value: Any) -> Any:
        encoder = entry.encoder
        if encoder is None:
            return value
        return encoder(value)

    def _decode_value(self, entry: _CacheEntry, payload: Any) -> Any:
        decoder = entry.decoder
        if decoder is None:
            return payload
        return decoder(payload)

    def _store_layer(
        self, name: str, entry: _CacheEntry, value: Any, *, layer_index: int
    ) -> None:
        layer = self._layers[layer_index]
        if layer_index == 0:
            payload = value
        else:
            try:
                payload = self._encode_value(entry, value)
            except Exception:  # pragma: no cover - defensive logging
                _logger.exception("cache encoding failed for %s", name)
                return
        try:
            layer.store(name, payload)
        except Exception:  # pragma: no cover - defensive logging
            _logger.exception(
                "cache layer store failed for %s on %s", name, layer.__class__.__name__
            )

    def _persist_layers(self, name: str, entry: _CacheEntry, value: Any) -> None:
        for index in range(len(self._layers)):
            self._store_layer(name, entry, value, layer_index=index)

    def _delete_from_layers(self, name: str) -> None:
        for layer in self._layers:
            try:
                layer.delete(name)
            except KeyError:
                continue
            except Exception:  # pragma: no cover - defensive logging
                _logger.exception(
                    "cache layer delete failed for %s on %s",
                    name,
                    layer.__class__.__name__,
                )

    def _load_from_layers(self, name: str, entry: _CacheEntry) -> Any:
        # Primary in-memory layer first for fast-path lookups.
        try:
            value = self._layers[0].load(name)
        except KeyError:
            value = None
        except Exception:  # pragma: no cover - defensive logging
            _logger.exception(
                "cache layer load failed for %s on %s",
                name,
                self._layers[0].__class__.__name__,
            )
            value = None
        if value is not None:
            return value

        # Fall back to slower layers and hydrate preceding caches on success.
        for index in range(1, len(self._layers)):
            layer = self._layers[index]
            try:
                payload = layer.load(name)
            except KeyError:
                continue
            except Exception:  # pragma: no cover - defensive logging
                _logger.exception(
                    "cache layer load failed for %s on %s",
                    name,
                    layer.__class__.__name__,
                )
                continue
            try:
                value = self._decode_value(entry, payload)
            except Exception:  # pragma: no cover - defensive logging
                _logger.exception("cache decoding failed for %s", name)
                continue
            if value is None:
                continue
            for prev_index in range(index):
                self._store_layer(name, entry, value, layer_index=prev_index)
            return value
        return None

    # ------------------------------------------------------------------
    # Metrics helpers

    def _ensure_metrics(self, name: str) -> _CacheMetrics:
        metrics = self._metrics.get(name)
        if metrics is None:
            with self._registry_lock:
                metrics = self._metrics.get(name)
                if metrics is None:
                    metrics = _CacheMetrics()
                    self._metrics[name] = metrics
        return metrics

    def increment_hit(
        self,
        name: str,
        *,
        amount: int = 1,
        duration: float | None = None,
    ) -> None:
        """Increase cache hit counters for `
        ame`` (optionally logging latency)."""

        metrics = self._ensure_metrics(name)
        with metrics.lock:
            metrics.hits += int(amount)
            if duration is not None:
                metrics.total_time += float(duration)
                metrics.timings += 1

    def increment_miss(
        self,
        name: str,
        *,
        amount: int = 1,
        duration: float | None = None,
    ) -> None:
        """Increase cache miss counters for `
        ame`` (optionally logging latency)."""

        metrics = self._ensure_metrics(name)
        with metrics.lock:
            metrics.misses += int(amount)
            if duration is not None:
                metrics.total_time += float(duration)
                metrics.timings += 1

    def increment_eviction(self, name: str, *, amount: int = 1) -> None:
        """Increase eviction count for cache `
        ame``."""

        metrics = self._ensure_metrics(name)
        with metrics.lock:
            metrics.evictions += int(amount)

    def record_timing(self, name: str, duration: float) -> None:
        """Accumulate ``duration`` into latency telemetry for `
        ame``."""

        metrics = self._ensure_metrics(name)
        with metrics.lock:
            metrics.total_time += float(duration)
            metrics.timings += 1

    @contextmanager
    def timer(self, name: str) -> TimingContext:
        """Context manager recording execution time for `
        ame``."""

        start = perf_counter()
        try:
            yield
        finally:
            self.record_timing(name, perf_counter() - start)

    def get_metrics(self, name: str) -> CacheStatistics:
        """Return a snapshot of telemetry collected for cache `
        ame``."""

        metrics = self._metrics.get(name)
        if metrics is None:
            return CacheStatistics()
        with metrics.lock:
            return metrics.snapshot()

    def iter_metrics(self) -> Iterator[tuple[str, CacheStatistics]]:
        """Yield ``(name, stats)`` pairs for every cache with telemetry."""

        with self._registry_lock:
            items = tuple(self._metrics.items())
        for name, metrics in items:
            with metrics.lock:
                yield name, metrics.snapshot()

    def aggregate_metrics(self) -> CacheStatistics:
        """Return aggregated telemetry statistics across all caches."""

        aggregate = CacheStatistics()
        for _, stats in self.iter_metrics():
            aggregate = aggregate.merge(stats)
        return aggregate

    def register_metrics_publisher(
        self, publisher: Callable[[str, CacheStatistics], None]
    ) -> None:
        """Register ``publisher`` to receive metrics snapshots on demand."""

        with self._registry_lock:
            self._metrics_publishers.append(publisher)

    def publish_metrics(
        self,
        *,
        publisher: Callable[[str, CacheStatistics], None] | None = None,
    ) -> None:
        """Send cached telemetry to ``publisher`` or all registered publishers."""

        if publisher is None:
            with self._registry_lock:
                publishers = tuple(self._metrics_publishers)
        else:
            publishers = (publisher,)
        if not publishers:
            return
        snapshot = tuple(self.iter_metrics())
        for emit in publishers:
            for name, stats in snapshot:
                try:
                    emit(name, stats)
                except Exception:  # pragma: no cover - defensive logging
                    _logger.exception("Cache metrics publisher failed for %s", name)

    def log_metrics(self, logger: logging.Logger, *, level: int = logging.INFO) -> None:
        """Emit cache metrics using ``logger`` for telemetry hooks."""

        for name, stats in self.iter_metrics():
            logger.log(
                level,
                "cache=%s hits=%d misses=%d evictions=%d timings=%d total_time=%.6f",
                name,
                stats.hits,
                stats.misses,
                stats.evictions,
                stats.timings,
                stats.total_time,
            )


try:
    from .init import get_logger as _get_logger
except ImportError:  # pragma: no cover - circular bootstrap fallback

    def _get_logger(name: str) -> logging.Logger:
        return logging.getLogger(name)


_logger = _get_logger(__name__)
get_logger = _get_logger


def prune_lock_mapping(
    cache: Mapping[K, Any] | MutableMapping[K, Any] | None,
    locks: MutableMapping[K, Any] | None,
) -> None:
    """Drop lock entries not present in ``cache``."""

    if locks is None:
        return
    if cache is None:
        cache_keys: set[K] = set()
    else:
        cache_keys = set(cache.keys())
    for key in list(locks.keys()):
        if key not in cache_keys:
            locks.pop(key, None)


# Compatibility names share the maintained LRU implementation.
InstrumentedLRUCache = UnifiedLRUCache
ManagedLRUCache = UnifiedLRUCache


@dataclass
class _SeedCacheState:
    """Container tracking the state for :class:`_SeedHashCache`."""

    cache: InstrumentedLRUCache[tuple[int, int], int] | None
    maxsize: int


@dataclass
class _CounterState(Generic[K]):
    """State bundle used by :class:`ScopedCounterCache`."""

    cache: InstrumentedLRUCache[K, int]
    locks: dict[K, threading.RLock]
    max_entries: int


# Key used to store the node set checksum in a graph's ``graph`` attribute.
NODE_SET_CHECKSUM_KEY = "_node_set_checksum_cache"

logger = _logger


# Helper to avoid importing ``tnfr.utils.init`` at module import time and keep
# circular dependencies at bay while still reusing the canonical numpy loader.
def _require_numpy():
    from ..mathematics.unified_numerical import np

    return np


# Graph key storing per-graph layer configuration overrides.
_GRAPH_CACHE_LAYERS_KEY = "_tnfr_cache_layers"

# Process-wide configuration for shared cache layers (Shelve/Redis).
_GLOBAL_CACHE_LAYER_CONFIG: dict[str, dict[str, Any]] = {}
_GLOBAL_CACHE_LOCK = threading.RLock()
_GLOBAL_CACHE_MANAGER: CacheManager | None = None

# Keys of cache entries dependent on the edge version. Any change to the edge
# set requires these to be dropped to avoid stale data.
EDGE_VERSION_CACHE_KEYS = ("_trig_version",)


def get_graph_version(graph: Any, key: str, default: int = 0) -> int:
    """Return integer version stored in ``graph`` under ``key``."""

    return int(graph.get(key, default))


def increment_graph_version(graph: Any, key: str) -> int:
    """Increment and store a version counter in ``graph`` under ``key``."""

    version = get_graph_version(graph, key) + 1
    graph[key] = version
    return version


def stable_json(obj: Any) -> str:
    """Return a JSON string with deterministic ordering for ``obj``."""

    from .io import json_dumps

    return json_dumps(
        obj,
        sort_keys=True,
        ensure_ascii=False,
        to_bytes=False,
    )


def _compute_node_repr_digest(obj: Any) -> tuple[str, bytes]:
    """Serialize one label without using node equality as a cache identity."""
    try:
        repr_ = stable_json(obj)
    except TypeError:
        repr_ = repr(obj)
    digest = hashlib.blake2b(repr_.encode("utf-8"), digest_size=16).digest()
    return repr_, digest


class _NodeIdentityKey:
    """Keep a label alive while its bounded digest entry uses object identity."""

    __slots__ = ("node",)

    def __init__(self, node: Any) -> None:
        self.node = node

    def __hash__(self) -> int:
        return id(self.node)

    def __eq__(self, other: object) -> bool:
        return isinstance(other, _NodeIdentityKey) and self.node is other.node


@lru_cache(maxsize=1024, typed=True)
def _cached_node_repr_digest(key: _NodeIdentityKey) -> tuple[str, bytes]:
    return _compute_node_repr_digest(key.node)


def _node_repr_digest(obj: Any) -> tuple[str, bytes]:
    """Return a stable digest cached for the actual node object.

    Identity is only an in-process cache key; it never enters serialization or
    digest bytes. Keeping the object in a bounded key prevents id reuse while
    separating equal labels with different nested types or representations.
    """
    return _cached_node_repr_digest(_NodeIdentityKey(obj))


# Preserve the existing functools cache diagnostics and uncached entry point.
_node_repr_digest.cache_info = _cached_node_repr_digest.cache_info  # type: ignore[attr-defined]
_node_repr_digest.cache_clear = _cached_node_repr_digest.cache_clear  # type: ignore[attr-defined]
_node_repr_digest.cache_parameters = _cached_node_repr_digest.cache_parameters  # type: ignore[attr-defined]
_node_repr_digest.__wrapped__ = _compute_node_repr_digest  # type: ignore[attr-defined]


def clear_node_repr_cache() -> None:
    """Clear cached node representations used for checksums."""

    _cached_node_repr_digest.cache_clear()


def configure_global_cache_layers(
    *,
    shelve: Mapping[str, Any] | None = None,
    redis: Mapping[str, Any] | None = None,
    replace: bool = False,
) -> None:
    """Update process-wide cache layer configuration.

    Parameters mirror the per-layer specifications accepted via graph metadata.
    Passing ``replace=True`` clears previous settings before applying new ones.
    Providing ``None`` for a layer while ``replace`` is true removes that layer
    from the configuration.
    """

    global _GLOBAL_CACHE_MANAGER
    with _GLOBAL_CACHE_LOCK:
        manager = _GLOBAL_CACHE_MANAGER
        _GLOBAL_CACHE_MANAGER = None
        if replace:
            _GLOBAL_CACHE_LAYER_CONFIG.clear()
        if shelve is not None:
            _GLOBAL_CACHE_LAYER_CONFIG["shelve"] = dict(shelve)
        elif replace:
            _GLOBAL_CACHE_LAYER_CONFIG.pop("shelve", None)
        if redis is not None:
            _GLOBAL_CACHE_LAYER_CONFIG["redis"] = dict(redis)
        elif replace:
            _GLOBAL_CACHE_LAYER_CONFIG.pop("redis", None)
    _close_cache_layers(manager)


def _resolve_layer_config(
    graph: MutableMapping[str, Any] | None,
) -> dict[str, dict[str, Any]]:
    resolved: dict[str, dict[str, Any]] = {}
    with _GLOBAL_CACHE_LOCK:
        for name, spec in _GLOBAL_CACHE_LAYER_CONFIG.items():
            resolved[name] = dict(spec)
    if graph is not None:
        overrides = graph.get(_GRAPH_CACHE_LAYERS_KEY)
        if isinstance(overrides, Mapping):
            for name in ("shelve", "redis"):
                layer_spec = overrides.get(name)
                if isinstance(layer_spec, Mapping):
                    resolved[name] = dict(layer_spec)
                elif layer_spec is None:
                    resolved.pop(name, None)
    return resolved


def _build_shelve_layer(spec: Mapping[str, Any]) -> ShelveCacheLayer | None:
    path = spec.get("path")
    if not path:
        return None
    flag = spec.get("flag", "c")
    protocol = spec.get("protocol")
    writeback = bool(spec.get("writeback", False))
    try:
        proto_arg = None if protocol is None else int(protocol)
    except (TypeError, ValueError):
        logger.warning("Invalid shelve protocol %r; falling back to default", protocol)
        proto_arg = None
    try:
        return ShelveCacheLayer(
            str(path),
            flag=str(flag),
            protocol=proto_arg,
            writeback=writeback,
        )
    except Exception:  # pragma: no cover - defensive logging
        logger.exception("Failed to initialise ShelveCacheLayer for path %r", path)
        return None


def _build_redis_layer(spec: Mapping[str, Any]) -> RedisCacheLayer | None:
    enabled = spec.get("enabled", True)
    if not enabled:
        return None
    namespace = spec.get("namespace")
    client = spec.get("client")
    if client is None:
        factory = spec.get("client_factory")
        if callable(factory):
            try:
                client = factory()
            except Exception:  # pragma: no cover - defensive logging
                logger.exception("Redis cache client factory failed")
                return None
        else:
            kwargs = spec.get("client_kwargs")
            if isinstance(kwargs, Mapping):
                try:  # pragma: no cover - optional dependency
                    import redis  # type: ignore
                except Exception:  # pragma: no cover - defensive logging
                    logger.exception(
                        "redis-py is required to build the configured Redis client"
                    )
                    return None
                try:
                    client = redis.Redis(**dict(kwargs))
                except Exception:  # pragma: no cover - defensive logging
                    logger.exception(
                        "Failed to initialise redis client with %r", kwargs
                    )
                    return None
    try:
        if namespace is None:
            return RedisCacheLayer(client=client)
        return RedisCacheLayer(client=client, namespace=str(namespace))
    except Exception:  # pragma: no cover - defensive logging
        logger.exception("Failed to initialise RedisCacheLayer")
        return None


def _build_cache_layers(config: Mapping[str, dict[str, Any]]) -> tuple[CacheLayer, ...]:
    layers: list[CacheLayer] = []
    shelve_spec = config.get("shelve")
    if isinstance(shelve_spec, Mapping):
        layer = _build_shelve_layer(shelve_spec)
        if layer is not None:
            layers.append(layer)
    redis_spec = config.get("redis")
    if isinstance(redis_spec, Mapping):
        layer = _build_redis_layer(redis_spec)
        if layer is not None:
            layers.append(layer)
    return tuple(layers)


def _close_cache_layers(manager: CacheManager | None) -> None:
    if manager is None:
        return
    layers = getattr(manager, "_layers", ())
    for layer in layers:
        close = getattr(layer, "close", None)
        if callable(close):
            try:
                close()
            except Exception:  # pragma: no cover - defensive logging
                logger.exception(
                    "Cache layer close failed for %s", layer.__class__.__name__
                )


def reset_global_cache_manager() -> None:
    """Dispose the shared cache manager and close attached layers."""

    global _GLOBAL_CACHE_MANAGER
    with _GLOBAL_CACHE_LOCK:
        manager = _GLOBAL_CACHE_MANAGER
        _GLOBAL_CACHE_MANAGER = None
    _close_cache_layers(manager)


def build_cache_manager(
    *,
    graph: MutableMapping[str, Any] | None = None,
    storage: MutableMapping[str, Any] | None = None,
    default_capacity: int | None = None,
    overrides: Mapping[str, int | None] | None = None,
) -> CacheManager:
    """Construct a :class:`CacheManager` honouring configured cache layers."""

    global _GLOBAL_CACHE_MANAGER
    if graph is None:
        with _GLOBAL_CACHE_LOCK:
            manager = _GLOBAL_CACHE_MANAGER
        if manager is not None:
            return manager

    layers = _build_cache_layers(_resolve_layer_config(graph))
    manager = CacheManager(
        storage,
        default_capacity=default_capacity,
        overrides=overrides,
        layers=layers,
    )

    if graph is None:
        with _GLOBAL_CACHE_LOCK:
            global_manager = _GLOBAL_CACHE_MANAGER
            if global_manager is None:
                _GLOBAL_CACHE_MANAGER = manager
                return manager
        _close_cache_layers(manager)
        return global_manager

    return manager


def _node_repr(n: Any) -> str:
    """Stable representation for node hashing and sorting."""

    return _node_repr_digest(n)[0]


def _iter_node_digests(nodes: Iterable[Any], *, presorted: bool) -> Iterable[bytes]:
    """Yield node digests in a deterministic order."""

    if presorted:
        for node in nodes:
            yield _node_repr_digest(node)[1]
    else:
        for _, digest in sorted(
            (_node_repr_digest(n) for n in nodes), key=lambda x: x[0]
        ):
            yield digest


def _same_node_snapshot(previous: tuple[Any, ...], current: tuple[Any, ...]) -> bool:
    """Compare actual node objects, preserving order and label representation.

    Equality alone conflates replacements such as ``1``, ``True``, and ``1.0``.
    These labels can have different serialization and sorting despite comparing
    equal, so a cached snapshot may be reused only for the same objects.
    """
    return previous is current or (
        len(previous) == len(current)
        and all(old is new for old, new in zip(previous, current))
    )


def _node_set_checksum_no_nodes(
    G: nx.Graph,
    graph: Any,
    *,
    presorted: bool,
    store: bool,
) -> str:
    """Compatibility helper using the same snapshot policy as explicit nodes."""
    return node_set_checksum(G, tuple(G.nodes()), presorted=presorted, store=store)


def node_set_checksum(
    G: nx.Graph,
    nodes: Iterable[Any] | None = None,
    *,
    presorted: bool = False,
    store: bool = True,
) -> str:
    """Return a BLAKE2b checksum of the selected nodes.

    A complete ordered snapshot and the sorting mode validate reuse. This
    detects replacement/reordering even when graph size is unchanged and
    prevents a subset or ``presorted`` result from contaminating another call.
    ``store=False`` removes the saved record, including on a cache hit.
    """

    graph = get_graph(G)
    if nodes is None:
        return _node_set_checksum_no_nodes(G, graph, presorted=presorted, store=store)

    snapshot = tuple(nodes)
    cached = graph.get(NODE_SET_CHECKSUM_KEY)
    if (
        isinstance(cached, tuple)
        and len(cached) == 4
        and _same_node_snapshot(cached[2], snapshot)
        and cached[3] is bool(presorted)
    ):
        checksum = cached[1]
    else:
        hasher = hashlib.blake2b(digest_size=16)
        for digest in _iter_node_digests(snapshot, presorted=presorted):
            hasher.update(digest)
        checksum = hasher.hexdigest()
    if store:
        # Replace legacy two/three-field entries even when the digest is the
        # same. Otherwise every later lookup would rebuild all node digests.
        graph[NODE_SET_CHECKSUM_KEY] = (
            checksum[:16],
            checksum,
            snapshot,
            bool(presorted),
        )
    else:
        graph.pop(NODE_SET_CHECKSUM_KEY, None)
    return checksum


@dataclass(slots=True)
class NodeCache:
    """Container for cached node data."""

    checksum: str
    nodes: tuple[Any, ...]
    sorted_nodes: tuple[Any, ...] | None = None
    idx: dict[Any, int] | None = None
    offset: dict[Any, int] | None = None
    owner: weakref.ReferenceType[Any] | None = field(
        default=None, repr=False, compare=False
    )
    offset_sorted: bool | None = None

    def __reduce__(self) -> tuple[Any, tuple[Any, ...]]:
        """Preserve legacy picklability while rebuilding runtime ownership."""
        return type(self), (
            self.checksum,
            self.nodes,
            self.sorted_nodes,
            self.idx,
            self.offset,
            None,
            getattr(self, "offset_sorted", None),
        )

    @property
    def n(self) -> int:
        return len(self.nodes)


def _update_node_cache(
    graph: Any,
    nodes: tuple[Any, ...],
    key: str,
    *,
    checksum: str,
    sorted_nodes: tuple[Any, ...] | None = None,
    owner: weakref.ReferenceType[Any] | None = None,
) -> None:
    """Store `
    odes`` and ``checksum`` in ``graph`` under ``key``."""

    graph[f"{key}_cache"] = NodeCache(
        checksum=checksum, nodes=nodes, sorted_nodes=sorted_nodes, owner=owner
    )
    graph[f"{key}_checksum"] = checksum


def _refresh_node_list_cache(
    G: nx.Graph,
    graph: Any,
    *,
    sort_nodes: bool,
    current_n: int,
    nodes: tuple[Any, ...] | None = None,
) -> tuple[Any, ...]:
    """Refresh the cached node list and return the nodes."""

    if nodes is None:
        nodes = tuple(G.nodes())
    checksum = node_set_checksum(G, nodes, store=True)
    sorted_nodes = tuple(sorted(nodes, key=_node_repr)) if sort_nodes else None
    _update_node_cache(
        graph,
        nodes,
        "_node_list",
        checksum=checksum,
        sorted_nodes=sorted_nodes,
        owner=weakref.ref(G),
    )
    graph["_node_list_len"] = current_n
    return nodes


def _cache_node_list(G: nx.Graph) -> tuple[Any, ...]:
    """Validate one ordered snapshot and reuse graph-owned derived maps.

    NetworkX permits mutations without TNFR version hooks, including filtered
    views and same-size replacement. Comparing the complete tuple preserves
    that detection without serializing/hash-sorting unchanged nodes.
    """

    graph = get_graph(G)
    cache: NodeCache | None = graph.get("_node_list_cache")
    snapshot = tuple(G.nodes())
    current_n = len(snapshot)
    dirty = bool(graph.pop("_node_list_dirty", False))
    sort_nodes = bool(graph.get("SORT_NODES", False))
    if (
        cache is None
        or getattr(cache, "owner", None) is None
        or cache.owner() is not G
        or not _same_node_snapshot(cache.nodes, snapshot)
        or dirty
    ):
        return _refresh_node_list_cache(
            G, graph, sort_nodes=sort_nodes, current_n=current_n, nodes=snapshot
        )
    graph.setdefault("_node_list_checksum", cache.checksum)
    graph["_node_list_len"] = current_n
    if sort_nodes and cache.sorted_nodes is None:
        cache.sorted_nodes = tuple(sorted(cache.nodes, key=_node_repr))
    return cache.nodes


def cached_node_list(G: nx.Graph) -> tuple[Any, ...]:
    """Public wrapper returning the cached node tuple for ``G``."""

    return _cache_node_list(G)


def _ensure_node_map(
    G: TNFRGraph,
    *,
    attrs: tuple[str, ...],
    sort: bool = False,
) -> dict[NodeId, int]:
    """Return cached node-to-index/offset mappings stored on ``NodeCache``."""

    graph = G.graph
    _cache_node_list(G)
    cache: NodeCache = graph["_node_list_cache"]

    missing = [
        attr
        for attr in attrs
        if getattr(cache, attr) is None
        or (attr == "offset" and cache.offset_sorted is not sort)
    ]
    if missing:
        if sort:
            nodes_opt = cache.sorted_nodes
            if nodes_opt is None:
                nodes_opt = tuple(sorted(cache.nodes, key=_node_repr))
                cache.sorted_nodes = nodes_opt
            nodes_seq = nodes_opt
        else:
            nodes_seq = cache.nodes
        node_ids = cast(tuple[NodeId, ...], nodes_seq)
        mappings: dict[str, dict[NodeId, int]] = {attr: {} for attr in missing}
        for idx, node in enumerate(node_ids):
            for attr in missing:
                mappings[attr][node] = idx
        for attr in missing:
            setattr(cache, attr, mappings[attr])
            if attr == "offset":
                cache.offset_sorted = sort
    return cast(dict[NodeId, int], getattr(cache, attrs[0]))


def ensure_node_index_map(G: TNFRGraph) -> dict[NodeId, int]:
    """Return cached node-to-index mapping for ``G``."""

    return _ensure_node_map(G, attrs=("idx",), sort=False)


def ensure_node_offset_map(G: TNFRGraph) -> dict[NodeId, int]:
    """Return cached node-to-offset mapping for ``G``."""

    sort = bool(G.graph.get("SORT_NODES", False))
    return _ensure_node_map(G, attrs=("offset",), sort=sort)


def _offset_scope_owner() -> tuple[int, Any]:
    """Keep a synchronous scope local even when an async context is copied."""
    try:
        task = asyncio.current_task()
    except RuntimeError:
        task = None
    return threading.get_ident(), task


@dataclass(slots=True)
class _NodeOffsetScope:
    graph: nx.Graph
    nodes: tuple[Any, ...]
    offsets: dict[Any, int]
    node_data: dict[Any, Any]
    storage: Any
    metadata: Any
    sort: bool
    owner: tuple[int, Any]
    active: bool = True

    def check(self, *, complete: bool = False) -> None:
        G = self.graph
        changed = (
            G._node is not self.storage
            or G.graph is not self.metadata
            or len(G._node) != len(self.nodes)
            or bool(G.graph.get("SORT_NODES", False)) is not self.sort
        )
        if complete and not changed:
            changed = not _same_node_snapshot(self.nodes, tuple(G.nodes())) or any(
                G._node[node] is not data for node, data in self.node_data.items()
            )
        if changed:
            raise RuntimeError(
                "stable_node_offsets: node order, storage, or SORT_NODES changed "
                "during the scope; earlier operations are not rolled back"
            )


_NODE_OFFSET_SCOPES: ContextVar[tuple[_NodeOffsetScope, ...]] = ContextVar(
    "tnfr_node_offset_scopes", default=()
)


def _scoped_node_offset(G: nx.Graph, node: NodeId) -> int | None:
    """Read an opted-in offset; unrelated/expired execution uses normal lookup."""
    for scope in reversed(_NODE_OFFSET_SCOPES.get()):
        if (
            scope.graph is not G
            or not scope.active
            or scope.owner != _offset_scope_owner()
        ):
            continue
        scope.check()
        # A replaced target must not consume its predecessor's offset before
        # the complete boundary check. Other same-size mutations are checked
        # at scope exit, under the caller's explicit stable-order contract.
        if node not in scope.offsets or G._node.get(node) is not scope.node_data[node]:
            raise RuntimeError(
                "stable_node_offsets: target node changed during the scope"
            )
        return scope.offsets[node]
    return None


@contextmanager
def stable_node_offsets(G: nx.Graph) -> Iterator[tuple[NodeId, ...]]:
    """Reuse jitter offsets during an explicitly owned synchronous traversal.

    The caller must have exclusive graph access and keep node membership,
    order, node attribute dictionaries, label representations and SORT_NODES
    unchanged until exit, including inside callbacks. Attribute *values* and
    edges may change. The yielded tuple retains graph insertion order; jitter
    offsets still follow SORT_NODES. Public NodeNX.offset always performs its
    normal complete validation, even inside this scope.

    Only ordinary NetworkX Graph/DiGraph/MultiGraph/MultiDiGraph instances are
    supported; views and subclasses retain their normal lookup path. A warm
    scope costs O(V) at its boundaries and O(1) per jitter offset for fixed
    nesting depth, using O(V) temporary storage. Offset selection is O(depth)
    for arbitrarily nested scopes. This is not O(1) arbitrary-mutation detection
    or a lock: final membership/order changes raise on exit, and earlier
    operations are not rolled back. Transient changes restored before exit
    are not tracked.

    Scopes nest, expire on exceptions, and are not inherited by other threads
    or async tasks. Do not yield/await while holding this synchronous scope.
    No snapshot is stored on the graph or carried into graph copies.

    Example::

        from tnfr.node import NodeNX
        from tnfr.operators.jitter import random_jitter
        from tnfr.utils.cache import stable_node_offsets

        with stable_node_offsets(graph) as nodes:
            for node_id in nodes:
                value = random_jitter(NodeNX.from_graph(graph, node_id), 0.1)
    """
    if type(G) not in (nx.Graph, nx.DiGraph, nx.MultiGraph, nx.MultiDiGraph) or hasattr(
        G, "_graph"
    ):
        raise TypeError(
            "stable_node_offsets requires an ordinary NetworkX graph, not a view/subclass"
        )
    owner = _offset_scope_owner()
    previous = _NODE_OFFSET_SCOPES.get()
    for outer in previous:
        if outer.graph is G and outer.active and outer.owner == owner:
            outer.check(complete=True)
    offsets = dict(ensure_node_offset_map(G))
    nodes = G.graph["_node_list_cache"].nodes
    scope = _NodeOffsetScope(
        G,
        nodes,
        offsets,
        {node: G._node[node] for node in nodes},
        G._node,
        G.graph,
        bool(G.graph.get("SORT_NODES", False)),
        owner,
    )
    token = _NODE_OFFSET_SCOPES.set((*previous, scope))
    try:
        try:
            yield nodes
        except BaseException as exc:
            try:
                scope.check(complete=True)
            except RuntimeError as violation:
                if hasattr(exc, "add_note"):  # Python 3.11+, preserving 3.10 support.
                    exc.add_note(str(violation))
            raise
        else:
            scope.check(complete=True)
    finally:
        scope.active = False
        _NODE_OFFSET_SCOPES.reset(token)


@dataclass
class EdgeCacheState:
    cache: MutableMapping[Hashable, Any]
    locks: defaultdict[Hashable, threading.RLock]
    max_entries: int | None
    dirty: bool = False


_GRAPH_CACHE_MANAGER_KEY = "_tnfr_cache_manager"
_GRAPH_CACHE_CONFIG_KEY = "_tnfr_cache_config"
DNFR_PREP_STATE_KEY = "_dnfr_prep_state"

# Rebuildable state owned by this cache/metrics infrastructure. Configuration
# (_cache_config, _tnfr_cache_config, _tnfr_cache_layers) is deliberately absent.
# Graph-state copies can omit these keys before copying persistent user data.
GRAPH_RUNTIME_CACHE_KEYS = frozenset(
    {
        _GRAPH_CACHE_MANAGER_KEY,
        "_edge_cache_manager",
        DNFR_PREP_STATE_KEY,
        "_dnfr_prep_cache",
        "_tnfr_change_tracker",
        "_trig_version",
        "_node_cache",
        "_node_cache_weak",
        "_node_list_cache",
        "_node_list_checksum",
        "_node_list_len",
        "_node_list_dirty",
        NODE_SET_CHECKSUM_KEY,
        "_dnfr_nodes_checksum",
    }
)

# Ephemeral graph cache management:
# ----------------------------------
# TNFR stores cache managers directly in each graph's `.graph` dictionary
# via _GRAPH_CACHE_MANAGER_KEY. This design inherently supports ephemeral
# graphs because:
#
# 1. **Automatic cleanup**: When an ephemeral graph object is garbage
#    collected, its `.graph` dict and all associated cache managers are
#    automatically released with it. No manual cleanup is required.
#
# 2. **Isolation**: Each graph has its own cache manager instance, preventing
#    cache pollution between unrelated graphs or temporary computations.
#
# 3. **No global state**: Unlike WeakValueDictionary-based global caches,
#    there's no shared cache registry that needs weak references to track
#    ephemeral graphs.
#
# For temporary or short-lived graphs (e.g., subgraphs, clones, simulation
# snapshots), simply let the graph go out of scope and Python's garbage
# collector will reclaim all associated caches. No special ephemeral flag
# or WeakValueDictionary is needed.
#
# Example ephemeral graph usage:
#   def process_subgraph(G, nodes):
#       H = G.subgraph(nodes).copy()  # Ephemeral graph
#       default_compute_delta_nfr(H)  # Creates temporary cache
#       return extract_metrics(H)
#       # H and its caches are GC'd when function returns


@dataclass(slots=True)
class DnfrPrepState:
    """State container coordinating ΔNFR preparation caches."""

    cache: DnfrCache
    cache_lock: threading.RLock
    vector_lock: threading.RLock


def _build_dnfr_prep_state(
    graph: MutableMapping[str, Any],
    previous: DnfrPrepState | None = None,
) -> DnfrPrepState:
    """Construct a :class:`DnfrPrepState` and mirror it on ``graph``."""

    cache_lock: threading.RLock
    vector_lock: threading.RLock
    if isinstance(previous, DnfrPrepState):
        cache_lock = previous.cache_lock
        vector_lock = previous.vector_lock
    else:
        cache_lock = threading.RLock()
        vector_lock = threading.RLock()
    state = DnfrPrepState(
        cache=new_dnfr_cache(),
        cache_lock=cache_lock,
        vector_lock=vector_lock,
    )
    graph["_dnfr_prep_cache"] = state.cache
    return state


def _coerce_dnfr_state(
    graph: MutableMapping[str, Any],
    current: Any,
) -> DnfrPrepState:
    """Return ``current`` normalised into :class:`DnfrPrepState`."""

    if isinstance(current, DnfrPrepState):
        graph["_dnfr_prep_cache"] = current.cache
        return current
    if isinstance(current, DnfrCache):
        state = DnfrPrepState(
            cache=current,
            cache_lock=threading.RLock(),
            vector_lock=threading.RLock(),
        )
        graph["_dnfr_prep_cache"] = current
        return state
    return _build_dnfr_prep_state(graph)


def _graph_cache_manager(graph: MutableMapping[str, Any]) -> CacheManager:
    manager = graph.get(_GRAPH_CACHE_MANAGER_KEY)
    # NetworkX copies graph metadata shallowly. Rebinding only the outer
    # EdgeCacheManager still shares the inner buffers and reset hooks.
    if (
        not isinstance(manager, CacheManager)
        or getattr(manager, "_graph_owner", None) is not graph
    ):
        manager = build_cache_manager(graph=graph, default_capacity=128)
        manager._graph_owner = graph
        graph[_GRAPH_CACHE_MANAGER_KEY] = manager
    config = graph.get(_GRAPH_CACHE_CONFIG_KEY)
    if isinstance(config, dict):
        manager.configure_from_mapping(config)

    def _dnfr_factory() -> DnfrPrepState:
        return _build_dnfr_prep_state(graph)

    def _dnfr_reset(current: Any) -> DnfrPrepState:
        if isinstance(current, DnfrPrepState):
            return _build_dnfr_prep_state(graph, current)
        return _build_dnfr_prep_state(graph)

    manager.register(
        DNFR_PREP_STATE_KEY,
        _dnfr_factory,
        reset=_dnfr_reset,
    )
    manager.update(
        DNFR_PREP_STATE_KEY,
        lambda current: _coerce_dnfr_state(graph, current),
    )
    return manager


def configure_graph_cache_limits(
    G: GraphLike | TNFRGraph | MutableMapping[str, Any],
    *,
    default_capacity: int | None | object = CacheManager._MISSING,
    overrides: Mapping[str, int | None] | None = None,
    replace_overrides: bool = False,
) -> CacheCapacityConfig:
    """Update cache capacity policy stored on ``G.graph``."""

    graph = get_graph(G)
    manager = _graph_cache_manager(graph)
    manager.configure(
        default_capacity=default_capacity,
        overrides=overrides,
        replace_overrides=replace_overrides,
    )
    snapshot = manager.export_config()
    graph[_GRAPH_CACHE_CONFIG_KEY] = {
        "default_capacity": snapshot.default_capacity,
        "overrides": dict(snapshot.overrides),
    }
    return snapshot


class EdgeCacheManager:
    """Coordinate cache storage and per-key locks for edge version caches."""

    _STATE_KEY = "_edge_version_state"

    def __init__(self, graph: MutableMapping[str, Any]) -> None:
        self.graph: MutableMapping[str, Any] = graph
        self._manager = _graph_cache_manager(graph)
        self._source_snapshot: Any = None
        self._source_references: tuple[Any, ...] = ()
        self._source_version: int | None = None

        def _encode_state(state: EdgeCacheState) -> Mapping[str, Any]:
            if not isinstance(state, EdgeCacheState):
                raise TypeError("EdgeCacheState expected")
            return {
                "max_entries": state.max_entries,
                "entries": list(state.cache.items()),
            }

        def _decode_state(payload: Any) -> EdgeCacheState:
            if isinstance(payload, EdgeCacheState):
                return payload
            if not isinstance(payload, Mapping):
                raise TypeError("invalid edge cache payload")
            max_entries = payload.get("max_entries")
            state = self._build_state(max_entries)
            for key, value in payload.get("entries", []):
                state.cache[key] = value
            state.dirty = False
            return state

        self._manager.register(
            self._STATE_KEY,
            self._default_state,
            reset=self._reset_state,
            encoder=_encode_state,
            decoder=_decode_state,
        )

    def record_hit(self) -> None:
        """Record a cache hit for telemetry."""

        self._manager.increment_hit(self._STATE_KEY)

    def record_miss(self, *, track_metrics: bool = True) -> None:
        """Record a cache miss for telemetry.

        When ``track_metrics`` is ``False`` the miss is acknowledged without
        mutating the aggregated metrics.
        """

        if track_metrics:
            self._manager.increment_miss(self._STATE_KEY)

    def record_eviction(self, *, track_metrics: bool = True) -> None:
        """Record cache eviction events for telemetry.

        When ``track_metrics`` is ``False`` the underlying metrics counter is
        left untouched while still signalling that an eviction occurred.
        """

        if track_metrics:
            self._manager.increment_eviction(self._STATE_KEY)

    def timer(self) -> TimingContext:
        """Return a timing context linked to this cache."""

        return self._manager.timer(self._STATE_KEY)

    def _default_state(self) -> EdgeCacheState:
        return self._build_state(None)

    def resolve_max_entries(self, max_entries: int | None | object) -> int | None:
        """Return effective capacity for the edge cache."""

        if max_entries is CacheManager._MISSING:
            return self._manager.get_capacity(self._STATE_KEY)
        return self._manager.get_capacity(
            self._STATE_KEY,
            requested=None if max_entries is None else int(max_entries),
            use_default=False,
        )

    def _build_state(self, max_entries: int | None) -> EdgeCacheState:
        locks: defaultdict[Hashable, threading.RLock] = defaultdict(threading.RLock)
        capacity = float("inf") if max_entries is None else int(max_entries)
        cache = InstrumentedLRUCache(
            capacity,
            manager=self._manager,
            metrics_key=self._STATE_KEY,
            locks=locks,
            count_overwrite_hit=False,
        )
        state = EdgeCacheState(cache=cache, locks=locks, max_entries=max_entries)

        def _on_eviction(key: Hashable, _: Any) -> None:
            self.record_eviction(track_metrics=False)
            locks.pop(key, None)
            state.dirty = True

        cache.set_eviction_callbacks(_on_eviction)
        return state

    def _ensure_state(
        self, state: EdgeCacheState | None, max_entries: int | None | object
    ) -> EdgeCacheState:
        target = self.resolve_max_entries(max_entries)
        if target is not None:
            target = int(target)
            if target < 0:
                raise TNFRValueError(
                    "max_entries must be non-negative or None",
                    context={"max_entries": target},
                )
        if not isinstance(state, EdgeCacheState) or state.max_entries != target:
            return self._build_state(target)
        return state

    def _reset_state(self, state: EdgeCacheState | None) -> EdgeCacheState:
        if isinstance(state, EdgeCacheState):
            state.cache.clear()
            state.dirty = False
            return state
        return self._build_state(None)

    def get_cache(
        self,
        max_entries: int | None | object,
        *,
        create: bool = True,
    ) -> EdgeCacheState | None:
        """Return the cache state for the manager's graph."""

        if not create:
            state = self._manager.peek(self._STATE_KEY)
            return state if isinstance(state, EdgeCacheState) else None

        state = self._manager.update(
            self._STATE_KEY,
            lambda current: self._ensure_state(current, max_entries),
        )
        if not isinstance(state, EdgeCacheState):
            raise RuntimeError("edge cache state failed to initialise")
        return state

    def flush_state(self, state: EdgeCacheState) -> None:
        """Persist ``state`` through the configured cache layers when dirty."""

        if not isinstance(state, EdgeCacheState) or not state.dirty:
            return
        self._manager.store(self._STATE_KEY, state)
        state.dirty = False

    def clear(self) -> None:
        """Reset cached data managed by this instance."""

        self._manager.clear(self._STATE_KEY)

    def ensure_current_source(self, G: nx.Graph) -> bool:
        """Observe direct NetworkX edits before reusing any edge artifact.

        This takes O(V+E) work for scalar edge channels. Explicit version
        updates still invalidate immediately; an observed edit under an
        unchanged version uses that same invalidation owner. Unreadable edge
        metadata disables reuse rather than imposing a new numerical domain.
        """
        observed = _edge_source_snapshot(G)
        version = get_graph_version(self.graph, "_edge_version")
        if observed is None:
            _reset_edge_caches(self.graph, G)
            self._source_snapshot = None
            self._source_references = ()
            self._source_version = version
            return False
        snapshot, references = observed
        if self._source_snapshot is None:
            # Restored/persistent cache entries have no live source witness.
            _reset_edge_caches(self.graph, G)
        elif snapshot != self._source_snapshot and version == self._source_version:
            increment_edge_version(G)
            version = get_graph_version(self.graph, "_edge_version")
        self._source_snapshot = snapshot
        self._source_references = references
        self._source_version = version
        return True


def _edge_source_snapshot(G: nx.Graph) -> tuple[Any, tuple[Any, ...]] | None:
    """Capture ordered support identity and the two declared edge channels.

    Labels and parallel keys are retained by identity, without sorting or
    calling their repr/equality. The shared structural signature preserves raw
    weight/length values, including malformed values a particular reader may
    legitimately ignore. Node values and other edge annotations remain the
    responsibility of each cache key or an explicit version update.
    """
    from ._structural_signature import (
        StructuralSignatureError,
        structural_proof_signature,
    )

    multiple = G.is_multigraph()
    references: list[Any] = []
    rows = []
    channels = []
    for node, adjacency in G.adjacency():
        references.append(node)
        neighbors = []
        for neighbor, data in adjacency.items():
            references.append(neighbor)
            entries = data.items() if multiple else ((None, data),)
            keys = []
            for key, attributes in entries:
                if multiple:
                    references.append(key)
                keys.append(id(key))
                if "weight" in attributes or "length" in attributes:
                    channels.append(
                        (
                            len(rows),
                            len(neighbors),
                            len(keys) - 1,
                            "weight" in attributes,
                            attributes.get("weight"),
                            "length" in attributes,
                            attributes.get("length"),
                        )
                    )
            neighbors.append((id(neighbor), tuple(keys)))
        rows.append((id(node), tuple(neighbors)))
    try:
        channel_signature = structural_proof_signature(
            tuple(channels), _retained_references=references
        )
    except StructuralSignatureError:
        return None
    return (
        (G.is_directed(), multiple, tuple(rows), channel_signature),
        tuple(references),
    )


def edge_version_cache(
    G: Any,
    key: Hashable,
    builder: Callable[[], T],
    *,
    max_entries: int | None | object = CacheManager._MISSING,
) -> T:
    """Return cached ``builder`` output tied to the edge source of ``G``.

    NetworkX views share their parent's metadata dictionary and may expose a
    different node/edge set. They compute fresh values instead of reading or
    writing that shared cache. Ordinary graph copies receive their own manager.
    NetworkX graphs also validate ordered support, node/key identity and raw
    weight/length channels, so direct edits cannot bypass invalidation. This
    check is O(V+E) for scalar edge channels. Other source dependencies require
    an appropriate key or explicit version update. Metadata-only/other graph
    adapters retain the explicit-version contract; concurrent mutation during
    a read is not supported.
    """

    if isinstance(G, nx.Graph) and getattr(G, "_graph", None) is not None:
        return builder()

    graph = get_graph(G)
    manager = graph.get("_edge_cache_manager")  # type: ignore[assignment]
    if not isinstance(manager, EdgeCacheManager) or manager.graph is not graph:
        manager = EdgeCacheManager(graph)
        graph["_edge_cache_manager"] = manager

    if isinstance(G, nx.Graph) and not manager.ensure_current_source(G):
        return builder()

    resolved = manager.resolve_max_entries(max_entries)
    if resolved == 0:
        return builder()

    state = manager.get_cache(resolved)
    if state is None:
        return builder()

    cache = state.cache
    locks = state.locks
    edge_version = get_graph_version(graph, "_edge_version")
    lock = locks[key]

    with lock:
        entry = cache.get(key)
        if entry is not None and entry[0] == edge_version:
            manager.record_hit()
            return entry[1]

    try:
        with manager.timer():
            value = builder()
    except (RuntimeError, ValueError) as exc:  # pragma: no cover - logging side effect
        logger.exception("edge_version_cache builder failed for %r: %s", key, exc)
        raise
    else:
        result = value
        with lock:
            entry = cache.get(key)
            if entry is not None:
                cached_version, cached_value = entry
                manager.record_miss()
                if cached_version == edge_version:
                    manager.record_hit()
                    return cached_value
                manager.record_eviction()
            cache[key] = (edge_version, value)
            state.dirty = True
            result = value
    if state.dirty:
        manager.flush_state(state)
    return result


def cached_nodes_and_A(
    G: nx.Graph,
    *,
    cache_size: int | None = 1,
    require_numpy: bool = False,
    prefer_sparse: bool = False,
    nodes: tuple[Any, ...] | None = None,
) -> tuple[tuple[Any, ...], Any]:
    """Return cached node order and binary outgoing-support adjacency for ΔNFR.

    Parallel edges count once, including zero-conductance edges: the phase,
    capacity and topology channels use unique outgoing neighbors. The separate
    EPI reducer reads live conductances, including parallel-edge sums.

    ``prefer_sparse`` skips matrix construction so ΔNFR can reuse its edge-index
    buffers without allocating dense adjacency.
    """

    if nodes is None:
        nodes = cached_node_list(G)
    nodes = tuple(nodes)
    graph = G.graph

    checksum = getattr(graph.get("_node_list_cache"), "checksum", None)
    if checksum is None:
        checksum = graph.get("_node_list_checksum")
    if checksum is None:
        node_set_cache = graph.get(NODE_SET_CHECKSUM_KEY)
        if isinstance(node_set_cache, tuple) and len(node_set_cache) >= 2:
            checksum = node_set_cache[1]
    if checksum is None:
        checksum = ""

    key = ("_dnfr", nodes, bool(prefer_sparse))
    graph["_dnfr_nodes_checksum"] = checksum

    def builder() -> tuple[tuple[Any, ...], Any]:
        np = _require_numpy()
        if np is None or prefer_sparse:
            return nodes, None
        A = nx.to_numpy_array(G, nodelist=nodes, weight=None, dtype=float)
        if G.is_multigraph():
            np.minimum(A, 1.0, out=A)
        return nodes, A

    nodes, A = edge_version_cache(G, key, builder, max_entries=cache_size)

    if require_numpy and A is None:
        raise RuntimeError("NumPy is required for adjacency caching")

    return nodes, A


def _reset_edge_caches(graph: Any, G: Any) -> None:
    """Clear caches affected by edge updates."""

    EdgeCacheManager(graph).clear()
    _graph_cache_manager(graph).clear(DNFR_PREP_STATE_KEY)
    mark_dnfr_prep_dirty(G)
    clear_node_repr_cache()
    for key in EDGE_VERSION_CACHE_KEYS:
        graph.pop(key, None)


def increment_edge_version(G: Any) -> None:
    """Increment the edge version counter in ``G.graph``."""

    graph = get_graph(G)
    increment_graph_version(graph, "_edge_version")
    _reset_edge_caches(graph, G)


@contextmanager
def edge_version_update(G: TNFRGraph) -> Iterator[None]:
    """Scope a batch of edge mutations."""

    increment_edge_version(G)
    try:
        yield
    finally:
        increment_edge_version(G)


class _SeedHashCache(MutableMapping[tuple[int, int], int]):
    """Mutable mapping proxy exposing a configurable LRU cache."""

    def __init__(
        self,
        *,
        manager: CacheManager | None = None,
        state_key: str = "seed_hash_cache",
        default_maxsize: int = 128,
    ) -> None:
        self._default_maxsize = int(default_maxsize)
        self._manager = manager or build_cache_manager(
            default_capacity=self._default_maxsize
        )
        self._state_key = state_key
        if not self._manager.has_override(self._state_key):
            self._manager.configure(overrides={self._state_key: self._default_maxsize})
        self._manager.register(
            self._state_key,
            self._create_state,
            reset=self._reset_state,
        )

    def _resolved_size(self, requested: int | None = None) -> int:
        size = self._manager.get_capacity(
            self._state_key,
            requested=requested,
            fallback=self._default_maxsize,
        )
        if size is None:
            return 0
        return int(size)

    def _create_state(self) -> _SeedCacheState:
        size = self._resolved_size()
        if size <= 0:
            return _SeedCacheState(cache=None, maxsize=0)
        return _SeedCacheState(
            cache=InstrumentedLRUCache(
                size,
                manager=self._manager,
                metrics_key=self._state_key,
            ),
            maxsize=size,
        )

    def _reset_state(self, state: _SeedCacheState | None) -> _SeedCacheState:
        return self._create_state()

    def _get_state(self, *, create: bool = True) -> _SeedCacheState | None:
        state = self._manager.get(self._state_key, create=create)
        if state is None:
            return None
        if not isinstance(state, _SeedCacheState):
            state = self._create_state()
            self._manager.store(self._state_key, state)
        return state

    def configure(self, maxsize: int) -> None:
        size = int(maxsize)
        if size < 0:
            raise TNFRValueError(
                "maxsize must be non-negative",
                context={"maxsize": size},
            )
        self._manager.configure(overrides={self._state_key: size})
        self._manager.update(self._state_key, lambda _: self._create_state())

    def __getitem__(self, key: tuple[int, int]) -> int:
        state = self._get_state()
        if state is None or state.cache is None:
            raise KeyError(key)
        value = state.cache[key]
        self._manager.increment_hit(self._state_key)
        return value

    def __setitem__(self, key: tuple[int, int], value: int) -> None:
        state = self._get_state()
        if state is not None and state.cache is not None:
            state.cache[key] = value

    def __delitem__(self, key: tuple[int, int]) -> None:
        state = self._get_state()
        if state is None or state.cache is None:
            raise KeyError(key)
        del state.cache[key]

    def __iter__(self) -> Iterator[tuple[int, int]]:
        state = self._get_state(create=False)
        if state is None or state.cache is None:
            return iter(())
        return iter(state.cache)

    def __len__(self) -> int:
        state = self._get_state(create=False)
        if state is None or state.cache is None:
            return 0
        return len(state.cache)

    def clear(self) -> None:  # type: ignore[override]
        self._manager.clear(self._state_key)

    @property
    def maxsize(self) -> int:
        state = self._get_state()
        return 0 if state is None else state.maxsize

    @property
    def enabled(self) -> bool:
        state = self._get_state(create=False)
        return bool(state and state.cache is not None)

    @property
    def data(self) -> InstrumentedLRUCache[tuple[int, int], int] | None:
        """Expose the underlying cache for diagnostics/tests."""

        state = self._get_state(create=False)
        return None if state is None else state.cache


class ScopedCounterCache(Generic[K]):
    """Thread-safe LRU cache storing monotonic counters by ``key``."""

    def __init__(
        self,
        name: str,
        max_entries: int | None = None,
        *,
        manager: CacheManager | None = None,
        default_max_entries: int = 128,
    ) -> None:
        self._name = name
        self._state_key = f"scoped_counter:{name}"
        self._default_max_entries = int(default_max_entries)
        requested = None if max_entries is None else int(max_entries)
        if requested is not None and requested < 0:
            raise TNFRValueError(
                "max_entries must be non-negative",
                context={"max_entries": requested},
            )
        self._manager = manager or build_cache_manager(
            default_capacity=self._default_max_entries
        )
        if not self._manager.has_override(self._state_key):
            fallback = requested
            if fallback is None:
                fallback = self._default_max_entries
            self._manager.configure(overrides={self._state_key: fallback})
        elif requested is not None:
            self._manager.configure(overrides={self._state_key: requested})
        self._manager.register(
            self._state_key,
            self._create_state,
            lock_factory=lambda: get_lock(name),
            reset=self._reset_state,
        )

    def _resolved_entries(self, requested: int | None = None) -> int:
        size = self._manager.get_capacity(
            self._state_key,
            requested=requested,
            fallback=self._default_max_entries,
        )
        if size is None:
            return 0
        return int(size)

    def _create_state(self, requested: int | None = None) -> _CounterState[K]:
        size = self._resolved_entries(requested)
        locks: dict[K, threading.RLock] = {}
        return _CounterState(
            cache=InstrumentedLRUCache(
                size,
                manager=self._manager,
                metrics_key=self._state_key,
                locks=locks,
            ),
            locks=locks,
            max_entries=size,
        )

    def _reset_state(self, state: _CounterState[K] | None) -> _CounterState[K]:
        return self._create_state()

    def _get_state(self) -> _CounterState[K]:
        state = self._manager.get(self._state_key)
        if not isinstance(state, _CounterState):
            state = self._create_state(0)
            self._manager.store(self._state_key, state)
        return state

    @property
    def lock(self) -> threading.Lock | threading.RLock:
        """Return the lock guarding access to the underlying cache."""

        return self._manager.get_lock(self._state_key)

    @property
    def max_entries(self) -> int:
        """Return the configured maximum number of cached entries."""

        return self._get_state().max_entries

    @property
    def cache(self) -> InstrumentedLRUCache[K, int]:
        """Expose the instrumented cache for inspection."""

        return self._get_state().cache

    @property
    def locks(self) -> dict[K, threading.RLock]:
        """Return the mapping of per-key locks tracked by the cache."""

        return self._get_state().locks

    def configure(self, *, force: bool = False, max_entries: int | None = None) -> None:
        """Resize or reset the cache keeping previous settings."""

        if max_entries is None:
            size = self._resolved_entries()
            update_policy = False
        else:
            size = int(max_entries)
            if size < 0:
                raise TNFRValueError(
                    "max_entries must be non-negative",
                    context={"max_entries": size},
                )
            update_policy = True

        def _update(state: _CounterState[K] | None) -> _CounterState[K]:
            if (
                not isinstance(state, _CounterState)
                or force
                or state.max_entries != size
            ):
                locks: dict[K, threading.RLock] = {}
                return _CounterState(
                    cache=InstrumentedLRUCache(
                        size,
                        manager=self._manager,
                        metrics_key=self._state_key,
                        locks=locks,
                    ),
                    locks=locks,
                    max_entries=size,
                )
            return cast(_CounterState[K], state)

        if update_policy:
            self._manager.configure(overrides={self._state_key: size})
        self._manager.update(self._state_key, _update)

    def clear(self) -> None:
        """Clear stored counters preserving ``max_entries``."""

        self.configure(force=True)

    def bump(self, key: K) -> int:
        """Return current counter for ``key`` and increment it atomically."""

        result: dict[str, Any] = {}

        def _update(state: _CounterState[K] | None) -> _CounterState[K]:
            if not isinstance(state, _CounterState):
                state = self._create_state(0)
            cache = state.cache
            locks = state.locks
            if key not in locks:
                locks[key] = threading.RLock()
            value = int(cache.get(key, 0))
            cache[key] = value + 1
            result["value"] = value
            return state

        self._manager.update(self._state_key, _update)
        return int(result.get("value", 0))

    def __len__(self) -> int:
        """Return the number of tracked counters."""

        return len(self.cache)


# ============================================================================
# Hierarchical Cache System (moved from caching/ for consolidation)
# ============================================================================

# class CacheLevel(Enum): ... (Moved to tnfr.types)


@dataclass
class CacheEntry:
    """Cache entry with metadata for intelligent invalidation and eviction.

    Attributes
    ----------
    value : Any
        The cached computation result.
    dependencies : set[str]
        set of structural properties this entry depends on. Used for
        selective invalidation. Examples: 'node_epi', 'node_vf', 'graph_topology'.
    timestamp : float
        Time when entry was created (from time.time()).
    access_count : int
        Number of times this entry has been accessed.
    computation_cost : float
        Estimated computational cost to regenerate this value. Higher cost
        entries are prioritized during eviction.
    size_bytes : int
        Estimated memory size in bytes.
    """

    value: Any
    dependencies: set[str]
    timestamp: float
    access_count: int = 0
    computation_cost: float = 1.0
    size_bytes: int = 0


class TNFRHierarchicalCache:
    """Hierarchical cache with dependency-aware selective invalidation.

    This cache system organizes entries by structural level and tracks
    dependencies to enable surgical invalidation. Only entries that depend
    on changed structural properties are evicted, preserving valid cached data.

    Internally uses ``CacheManager`` for unified cache management, metrics,
    and telemetry integration with the rest of TNFR.

    **Implementation characteristics** (v2):
    - Direct cache references bypass CacheManager dispatch on the hot path
    - Lazy persistence batches writes to persistent layers
    - type-based size estimation caching reduces memory tracking overhead
    - Dependency change detection avoids redundant updates
    - Batched invalidation reduces persistence operations

    **TNFR Compliance**:
    - Maintains §3.8 Controlled Determinism through consistent cache behavior
    - Supports §3.4 Operator Closure via dependency tracking

    Parameters
    ----------
    max_memory_mb : int, default: 512
        Maximum memory usage in megabytes before eviction starts.
    enable_metrics : bool, default: True
        Whether to track cache hit/miss metrics for telemetry.
    cache_manager : CacheManager, optional
        Existing CacheManager to use. If None, creates a new one.
    lazy_persistence : bool, default: True
        Enable lazy write-behind caching for persistent layers. When True,
        cache modifications are batched and written on flush or critical operations.
        This significantly improves write performance at the cost of potential
        data loss on ungraceful termination. set to False for immediate consistency.

    Attributes
    ----------
    hits : int
        Number of successful cache retrievals.
    misses : int
        Number of cache misses.
    evictions : int
        Number of entries evicted due to memory pressure.
    invalidations : int
        Number of entries invalidated due to dependency changes.

    Examples
    --------
    >>> cache = TNFRHierarchicalCache(max_memory_mb=128)
    >>> # Cache a derived metric with dependencies
    >>> cache.set(
    ...     "coherence_global",
    ...     0.95,
    ...     CacheLevel.DERIVED_METRICS,
    ...     dependencies={'graph_topology', 'all_node_vf'},
    ...     computation_cost=100.0
    ... )
    >>> cache.get("coherence_global", CacheLevel.DERIVED_METRICS)
    0.95
    >>> # Invalidate when topology changes
    >>> cache.invalidate_by_dependency('graph_topology')
    >>> cache.get("coherence_global", CacheLevel.DERIVED_METRICS)

    >>> # Flush lazy writes to persistent storage
    >>> cache.flush_dirty_caches()

    """

    def __init__(
        self,
        max_memory_mb: int = 512,
        enable_metrics: bool = True,
        cache_manager: CacheManager | None = None,
        lazy_persistence: bool = True,
    ):
        # Use provided CacheManager or create new one
        if cache_manager is None:
            # Estimate entries per MB (rough heuristic: ~100 entries per MB)
            default_capacity = max(32, int(max_memory_mb * 100 / len(CacheLevel)))
            cache_manager = CacheManager(
                storage={},
                default_capacity=default_capacity,
            )

        self._manager = cache_manager
        self._max_memory = max_memory_mb * 1024 * 1024
        self._current_memory = 0
        self._enable_metrics = enable_metrics
        self._lazy_persistence = lazy_persistence

        # Dependency tracking (remains in hierarchical cache)
        self._dependencies: dict[str, set[tuple[CacheLevel, str]]] = defaultdict(set)

        # Register a cache for each level in the CacheManager
        self._level_cache_names: dict[CacheLevel, str] = {}
        # OPTIMIZATION: Direct cache references to avoid CacheManager overhead on hot path
        self._direct_caches: dict[CacheLevel, dict[str, CacheEntry]] = {}

        for level in CacheLevel:
            cache_name = f"hierarchical_{level.value}"
            self._level_cache_names[level] = cache_name

            # Simple factory returning empty dict for each cache level
            self._manager.register(
                cache_name,
                factory=lambda: {},
                create=True,
            )

            # Store direct reference for fast access
            self._direct_caches[level] = self._manager.get(cache_name)

        # OPTIMIZATION: Track dirty caches for batched persistence
        self._dirty_levels: set[CacheLevel] = set()

        # OPTIMIZATION: type-based size estimation cache
        self._size_cache: dict[type, int] = {}

        # Metrics (tracked locally for backward compatibility)
        self.hits = 0
        self.misses = 0
        self.evictions = 0
        self.invalidations = 0

    @property
    def _caches(self) -> dict[CacheLevel, dict[str, CacheEntry]]:
        """Provide backward compatibility for accessing internal caches.

        This property returns a view of the caches stored in the CacheManager,
        maintaining compatibility with code that directly accessed the old
        _caches attribute.

        Note: Uses direct cache references for performance.
        """
        return self._direct_caches

    def get(self, key: str, level: CacheLevel) -> Any | None:
        """Retrieve value from cache if it exists and is valid.

        Parameters
        ----------
        key : str
            Cache key identifying the entry.
        level : CacheLevel
            Cache level to search in.

        Returns
        -------
        Any or None
            The cached value if found, None otherwise.

        Examples
        --------
        >>> cache = TNFRHierarchicalCache()
        >>> cache.set("key1", 42, CacheLevel.TEMPORARY, dependencies=set())
        >>> cache.get("key1", CacheLevel.TEMPORARY)
        42
        >>> cache.get("missing", CacheLevel.TEMPORARY)

        """
        # OPTIMIZATION: Use direct cache reference to avoid CacheManager overhead
        level_cache = self._direct_caches[level]

        if key in level_cache:
            entry = level_cache[key]
            entry.access_count += 1
            if self._enable_metrics:
                self.hits += 1
                # Only update manager metrics if not in lazy mode
                if not self._lazy_persistence:
                    cache_name = self._level_cache_names[level]
                    self._manager.increment_hit(cache_name)
            return entry.value

        if self._enable_metrics:
            self.misses += 1
            if not self._lazy_persistence:
                cache_name = self._level_cache_names[level]
                self._manager.increment_miss(cache_name)
        return None

    def set(
        self,
        key: str,
        value: Any,
        level: CacheLevel,
        dependencies: set[str],
        computation_cost: float = 1.0,
    ) -> None:
        """Store value in cache with dependency metadata.

        Parameters
        ----------
        key : str
            Unique identifier for this cache entry.
        value : Any
            The value to cache.
        level : CacheLevel
            Which cache level to store in.
        dependencies : set[str]
            set of structural properties this value depends on.
        computation_cost : float, default: 1.0
            Estimated cost to recompute this value. Used for eviction priority.

        Examples
        --------
        >>> cache = TNFRHierarchicalCache()
        >>> cache.set(
        ...     "si_node_5",
        ...     0.87,
        ...     CacheLevel.DERIVED_METRICS,
        ...     dependencies={'node_vf_5', 'node_phase_5'},
        ...     computation_cost=5.0
        ... )
        """
        # OPTIMIZATION: Use direct cache reference
        level_cache = self._direct_caches[level]

        # OPTIMIZATION: Lazy size estimation - estimate size once
        estimated_size = self._estimate_size_fast(value)

        # Check if we need to evict
        if self._current_memory + estimated_size > self._max_memory:
            self._evict_lru(estimated_size)

        # Create entry
        entry = CacheEntry(
            value=value,
            dependencies=dependencies.copy(),
            timestamp=time.time(),
            computation_cost=computation_cost,
            size_bytes=estimated_size,
        )

        # Remove old entry if exists
        old_dependencies: set[str] | None = None
        if key in level_cache:
            old_entry = level_cache[key]
            self._current_memory -= old_entry.size_bytes
            old_dependencies = old_entry.dependencies
            # OPTIMIZATION: Only clean up dependencies if they changed
            if old_dependencies != dependencies:
                for dep in old_dependencies:
                    if dep in self._dependencies:
                        self._dependencies[dep].discard((level, key))

        # Store entry (direct modification, no manager overhead)
        level_cache[key] = entry
        self._current_memory += estimated_size

        # OPTIMIZATION: Register dependencies only if new or changed
        if old_dependencies is None or old_dependencies != dependencies:
            for dep in dependencies:
                self._dependencies[dep].add((level, key))

        # OPTIMIZATION: Mark level as dirty for lazy persistence
        if self._lazy_persistence:
            self._dirty_levels.add(level)
        else:
            # Immediate persistence (backward compatible)
            cache_name = self._level_cache_names[level]
            self._manager.store(cache_name, level_cache)

    def invalidate_by_dependency(self, dependency: str) -> int:
        """Invalidate all cache entries that depend on a structural property.

        This implements selective invalidation: only entries that explicitly
        depend on the changed property are removed, preserving unaffected caches.

        Parameters
        ----------
        dependency : str
            The structural property that changed (e.g., 'graph_topology',
            'node_epi_5', 'all_node_vf').

        Returns
        -------
        int
            Number of entries invalidated.

        Examples
        --------
        >>> cache = TNFRHierarchicalCache()
        >>> cache.set("key1", 1, CacheLevel.TEMPORARY, {'dep1', 'dep2'})
        >>> cache.set("key2", 2, CacheLevel.TEMPORARY, {'dep2'})
        >>> cache.invalidate_by_dependency('dep1')  # Only invalidates key1
        1
        >>> cache.get("key1", CacheLevel.TEMPORARY)  # None

        >>> cache.get("key2", CacheLevel.TEMPORARY)  # Still cached
        2
        """
        count = 0
        if dependency in self._dependencies:
            entries_to_remove = list(self._dependencies[dependency])
            invalidated_levels: set[CacheLevel] = set()

            for level, key in entries_to_remove:
                # OPTIMIZATION: Use direct cache reference
                level_cache = self._direct_caches[level]

                if key in level_cache:
                    entry = level_cache[key]
                    self._current_memory -= entry.size_bytes
                    del level_cache[key]
                    count += 1
                    invalidated_levels.add(level)

                    # Clean up all dependency references for this entry
                    for dep in entry.dependencies:
                        if dep in self._dependencies:
                            self._dependencies[dep].discard((level, key))

            # Clean up the dependency key itself
            del self._dependencies[dependency]

            # OPTIMIZATION: Batch persist invalidated levels
            if self._lazy_persistence:
                self._dirty_levels.update(invalidated_levels)
            else:
                for level in invalidated_levels:
                    cache_name = self._level_cache_names[level]
                    level_cache = self._direct_caches[level]
                    self._manager.store(cache_name, level_cache)

        if self._enable_metrics:
            self.invalidations += count

        return count

    def invalidate_level(self, level: CacheLevel) -> int:
        """Invalidate all entries in a specific cache level.

        Parameters
        ----------
        level : CacheLevel
            The cache level to clear.

        Returns
        -------
        int
            Number of entries invalidated.
        """
        # OPTIMIZATION: Use direct cache reference
        level_cache = self._direct_caches[level]
        count = len(level_cache)

        # Clean up dependencies
        for key, entry in level_cache.items():
            self._current_memory -= entry.size_bytes
            for dep in entry.dependencies:
                if dep in self._dependencies:
                    self._dependencies[dep].discard((level, key))

        level_cache.clear()

        # OPTIMIZATION: Batch persist if in lazy mode
        if self._lazy_persistence:
            self._dirty_levels.add(level)
        else:
            cache_name = self._level_cache_names[level]
            self._manager.store(cache_name, level_cache)

        if self._enable_metrics:
            self.invalidations += count

        return count

    def clear(self) -> None:
        """Clear all cache levels and reset metrics."""
        for level in CacheLevel:
            # OPTIMIZATION: Clear direct cache and update manager
            level_cache = self._direct_caches[level]
            level_cache.clear()
            cache_name = self._level_cache_names[level]
            self._manager.store(cache_name, level_cache)

        self._dependencies.clear()
        self._current_memory = 0
        self._dirty_levels.clear()

        # Always reset metrics regardless of _enable_metrics
        self.hits = 0
        self.misses = 0
        self.evictions = 0
        self.invalidations = 0

    def get_stats(self) -> dict[str, Any]:
        """Get cache statistics for telemetry.

        Returns
        -------
        dict[str, Any]
            Dictionary containing:
            - hits: Number of cache hits
            - misses: Number of cache misses
            - hit_rate: Ratio of hits to total accesses
            - evictions: Number of evictions
            - invalidations: Number of invalidations
            - memory_used_mb: Current memory usage in MB
            - memory_limit_mb: Memory limit in MB
            - entry_counts: Number of entries per level
        """
        total_accesses = self.hits + self.misses
        hit_rate = self.hits / total_accesses if total_accesses > 0 else 0.0

        entry_counts = {}
        for level in CacheLevel:
            # OPTIMIZATION: Use direct cache reference
            level_cache = self._direct_caches[level]
            entry_counts[level.value] = len(level_cache)

        return {
            "hits": self.hits,
            "misses": self.misses,
            "hit_rate": hit_rate,
            "evictions": self.evictions,
            "invalidations": self.invalidations,
            "memory_used_mb": self._current_memory / (1024 * 1024),
            "memory_limit_mb": self._max_memory / (1024 * 1024),
            "entry_counts": entry_counts,
        }

    def memory_profile(self) -> dict[str, Any]:
        """Return retained cache bytes by declared ownership level.

        This is accounting for ``CacheEntry.size_bytes`` retained by this
        instance, not a process-RSS measurement. It supports cache ownership
        audits without conflating independent graph or backend cache lifetimes.
        """
        by_level = {
            level.value: sum(entry.size_bytes for entry in cache.values())
            for level, cache in self._direct_caches.items()
        }
        return {
            "retained_bytes": sum(by_level.values()),
            "retained_bytes_by_level": by_level,
            "memory_limit_bytes": self._max_memory,
        }

    def _estimate_size(self, value: Any) -> int:
        """Estimate memory size of a value in bytes.

        Uses sys.getsizeof for a rough estimate. For complex objects,
        this may underestimate the true memory usage.
        """
        try:
            return sys.getsizeof(value)
        except (TypeError, AttributeError):
            # Fallback for objects that don't support getsizeof
            return 64  # Default estimate

    def _estimate_size_fast(self, value: Any) -> int:
        """Optimized size estimation with type-based caching.

        For common types, uses cached size estimates to avoid repeated
        sys.getsizeof() calls. Falls back to full estimation for complex types.
        """
        value_type = type(value)

        # Check if we have a cached size for this type
        if value_type in self._size_cache:
            # For simple immutable types, use cached size
            if value_type in (int, float, bool, type(None)):
                return self._size_cache[value_type]
            # For strings, estimate based on length
            if value_type is str:
                base_size = self._size_cache[value_type]
                return base_size + len(value)

        # Calculate size and cache for simple types
        size = self._estimate_size(value)
        if value_type in (int, float, bool, type(None)):
            self._size_cache[value_type] = size
        elif value_type is str:
            # Cache base size for strings
            if value_type not in self._size_cache:
                self._size_cache[value_type] = sys.getsizeof("")

        return size

    def flush_dirty_caches(self) -> None:
        """Flush dirty caches to persistent layers.

        In lazy persistence mode, this method writes accumulated changes
        to the CacheManager's persistent layers. This reduces write overhead
        by batching updates.
        """
        if not self._dirty_levels:
            return

        for level in self._dirty_levels:
            cache_name = self._level_cache_names[level]
            level_cache = self._direct_caches[level]
            self._manager.store(cache_name, level_cache)

        self._dirty_levels.clear()

    def _evict_lru(self, needed_space: int) -> None:
        """Evict least valuable entries until enough space is freed.

        Value is determined by: (access_count + 1) * computation_cost.
        Lower values are evicted first (low access, low cost to recompute).

        OPTIMIZED: Uses direct cache references and incremental eviction.
        """
        # OPTIMIZATION: Collect entries with direct cache access (no manager overhead)
        all_entries: list[tuple[float, CacheLevel, str, CacheEntry]] = []
        for level in CacheLevel:
            level_cache = self._direct_caches[level]
            for key, entry in level_cache.items():
                # Priority = (access_count + 1) * computation_cost
                # Higher priority = keep longer
                # Add 1 to access_count to avoid zero priority
                priority = (entry.access_count + 1) * entry.computation_cost
                all_entries.append((priority, level, key, entry))

        # Sort by priority (ascending - lowest priority first)
        all_entries.sort(key=lambda x: x[0])

        freed_space = 0
        evicted_levels: set[CacheLevel] = set()

        for priority, level, key, entry in all_entries:
            if freed_space >= needed_space:
                break

            # OPTIMIZATION: Remove entry directly from cache
            level_cache = self._direct_caches[level]
            if key in level_cache:
                del level_cache[key]
                freed_space += entry.size_bytes
                self._current_memory -= entry.size_bytes
                evicted_levels.add(level)

                # Clean up dependencies
                for dep in entry.dependencies:
                    if dep in self._dependencies:
                        self._dependencies[dep].discard((level, key))

                if self._enable_metrics:
                    self.evictions += 1
                    if not self._lazy_persistence:
                        cache_name = self._level_cache_names[level]
                        self._manager.increment_eviction(cache_name)

        # OPTIMIZATION: Batch persist evicted levels if in lazy mode
        if self._lazy_persistence:
            self._dirty_levels.update(evicted_levels)
        else:
            # Immediate persistence
            for level in evicted_levels:
                cache_name = self._level_cache_names[level]
                level_cache = self._direct_caches[level]
                self._manager.store(cache_name, level_cache)


# ============================================================================
# Cache Decorators (moved from caching/decorators.py for consolidation)
# ============================================================================

# Global cache instance shared across all decorated functions
_global_cache: TNFRHierarchicalCache | None = None

F = TypeVar("F", bound=Callable[..., Any])


def get_global_cache() -> TNFRHierarchicalCache:
    """Get or create the global TNFR cache instance.

    Returns
    -------
    TNFRHierarchicalCache
        The global cache instance.
    """
    global _global_cache
    if _global_cache is None:
        _global_cache = TNFRHierarchicalCache(max_memory_mb=512)
    return _global_cache


def set_global_cache(cache: TNFRHierarchicalCache | None) -> None:
    """set the global cache instance.

    Parameters
    ----------
    cache : TNFRHierarchicalCache or None
        The cache instance to use globally, or None to reset to default.
    """
    global _global_cache
    _global_cache = cache


def reset_global_cache() -> None:
    """Reset the global cache instance to None.

    The next call to get_global_cache() will create a fresh instance.
    """
    global _global_cache
    _global_cache = None


def _generate_cache_key(
    func_name: str,
    args: tuple[Any, ...],
    kwargs: dict[str, Any],
) -> str:
    """Generate deterministic cache key from function and arguments.

    Parameters
    ----------
    func_name : str
        Name of the function being cached.
    args : tuple
        Positional arguments.
    kwargs : dict
        Keyword arguments.

    Returns
    -------
    str
        Cache key string.

    Notes
    -----
    Uses MD5 for hashing (acceptable for cache keys, not security).
    Graph objects use id() which is session-specific - cache is cleared
    between sessions, so this is deterministic within a session.
    """
    # Build key components
    key_parts = [func_name]

    def argument_key(value: Any) -> str:
        # Signature binding turns positional graphs into keyword arguments;
        # both forms must retain identity instead of NetworkX's summary str.
        if hasattr(value, "graph"):
            return f"graph:{id(value)}"
        return repr(value)

    # Add positional args
    for arg in args:
        key_parts.append(argument_key(arg))

    # Add keyword args (sorted for consistency)
    for k in sorted(kwargs.keys()):
        v = kwargs[k]
        key_parts.append(f"{k}={argument_key(v)}")

    # Create deterministic hash (MD5 is acceptable for non-security cache keys)
    key_str = repr(key_parts)
    return hashlib.md5(key_str.encode(), usedforsecurity=False).hexdigest()


def _extract_graph_from_args(args: tuple, kwargs: dict) -> Any | None:
    """Find the first explicit graph argument, independent of its name."""
    for value in (*args, *kwargs.values()):
        if hasattr(value, "graph") and callable(getattr(value, "nodes", None)):
            return value
    return None


# Sentinel distinguishing "alias key absent" from "value is None" when
# hashing node-attribute dependencies for cache keys.
_DEP_HASH_MISSING = object()


def _dependency_alias_keys(dep: str) -> tuple[str, ...]:
    """Map a `
    ode_*`` cache dependency to its canonical alias keys.

        The canonical field writer (:func:`tnfr.alias.set_attr`) stores each
        field under the FIRST alias of its tuple, which is the Greek/canonical
        key (e.g. ``'ΔNFR'``, ``'νf'``, ``'EPI'``) — NOT the English dependency
        token.  The cache-key dependency hash MUST therefore resolve through the
        alias tuple; reading a hardcoded English name (``data='delta_nfr'``)
        silently yields ``None`` for every node, making the key blind to the
        field and returning stale results (the bug this maps around).
    """
    try:
        from ..constants.aliases import (
            ALIAS_DEPI,
            ALIAS_DNFR,
            ALIAS_EPI,
            ALIAS_THETA,
            ALIAS_VF,
        )
    except Exception:
        return ()
    mapping = {
        "node_phase": ALIAS_THETA,
        "node_dnfr": ALIAS_DNFR,
        "node_depi": ALIAS_DEPI,
        "node_epi": ALIAS_EPI,
        "node_vf": ALIAS_VF,
    }
    keys = mapping.get(dep)
    return tuple(keys) if keys else ()


def _compute_dependency_hash(graph: Any, dependencies: set[str]) -> str:
    """Compute a hash of the graph properties specified in dependencies.

    This ensures that cache keys reflect the current state of mutable
    node attributes like phase and delta_nfr, and that different subgraphs
    (with different node sets or edge sets) produce distinct keys when
    ``graph_topology`` is declared as a dependency. Precision-aware consumers
    declare ``precision_mode`` to separate results across global mode changes.
    """
    if not dependencies:
        return ""

    node_deps = {d for d in dependencies if d.startswith("node_")}
    has_topology = "graph_topology" in dependencies
    has_precision = "precision_mode" in dependencies
    if not node_deps and not has_topology and not has_precision:
        return ""

    hasher = hashlib.md5(usedforsecurity=False)

    # Check if graph is a NetworkX graph
    is_nx = hasattr(graph, "nodes") and callable(graph.nodes)

    if not is_nx and not has_precision:
        return ""

    def update_record(record: Any) -> None:
        # Delimit records structurally: [1, 23] must not hash like [12, 3].
        encoded = repr(record).encode("utf-8")
        hasher.update(len(encoded).to_bytes(8, "big"))
        hasher.update(encoded)

    if has_precision:
        from ..config.precision_modes import get_precision_mode

        update_record(("precision_mode", get_precision_mode()))

    if not is_nx:
        return hasher.hexdigest()

    # Include labelled topology plus both edge channels. ``weight`` is the
    # transport conductance and legacy path-length fallback; an explicit
    # ``length`` overrides it only for structural-potential geometry.
    if has_topology:
        try:
            directed = graph.is_directed()
            multigraph = graph.is_multigraph()
            update_record(("graph", directed, multigraph))
            update_record(("nodes", sorted(_node_repr(node) for node in graph)))
            edges = (
                graph.edges(keys=True, data=True)
                if multigraph
                else ((u, v, None, data) for u, v, data in graph.edges(data=True))
            )
            edge_records = []
            for u, v, key, data in edges:
                endpoints = (_node_repr(u), _node_repr(v))
                if not directed:
                    endpoints = tuple(sorted(endpoints))
                edge_records.append(
                    (
                        *endpoints,
                        repr(key),
                        repr(data.get("weight", 1.0)),
                        repr(data.get("length", None)),
                    )
                )
            update_record(("edges", sorted(edge_records)))
        except Exception:
            pass

    # Sort dependencies for determinism
    sorted_deps = sorted(node_deps)

    for dep in sorted_deps:
        alias_keys = _dependency_alias_keys(dep)
        if not alias_keys:
            continue
        try:
            # Alias precedence is per node, just as in the field readers.
            # Mixed canonical and legacy keys must all participate.
            items = []
            for node, data in graph.nodes(data=True):
                value = next(
                    (data[key] for key in alias_keys if key in data),
                    _DEP_HASH_MISSING,
                )
                items.append(
                    (
                        _node_repr(node),
                        value is not _DEP_HASH_MISSING,
                        repr(value) if value is not _DEP_HASH_MISSING else "",
                    )
                )
            update_record((dep, sorted(items)))
        except Exception:
            # If graph doesn't support this, skip
            pass

    return hasher.hexdigest()


def cache_tnfr_computation(
    level: CacheLevel,
    dependencies: set[str],
    cost_estimator: Callable[..., float] | None = None,
    cache_instance: TNFRHierarchicalCache | None = None,
) -> Callable[[F], F]:
    """Decorator for automatic caching of TNFR computations.

    Caches function results based on arguments and invalidates when
    dependencies change. Transparently integrates with existing functions.

    Parameters
    ----------
    level : CacheLevel
        Cache level for storing results.
    dependencies : set[str]
        set of structural properties this computation depends on.
        Examples: {'graph_topology', 'node_epi', 'node_vf', 'node_phase'}.
        Include 'precision_mode' when computation reads global precision state.
    cost_estimator : callable, optional
        Function that takes same arguments as decorated function and returns
        estimated computational cost as float. Used for eviction priority.
    cache_instance : TNFRHierarchicalCache, optional
        Specific cache instance to use. If None, uses global cache.

    Returns
    -------
    callable
        Decorated function with caching.

    Examples
    --------
    >>> from tnfr.cache import cache_tnfr_computation, CacheLevel
    >>> @cache_tnfr_computation(
    ...     level=CacheLevel.DERIVED_METRICS,
    ...     dependencies={'node_vf', 'node_phase'},
    ...     cost_estimator=lambda graph, node_id: len(list(graph.neighbors(node_id)))
    ... )
    ... def compute_metric(graph, node_id):
    ...     # Expensive computation
    ...     return 0.85

    With custom cache instance:

    >>> from tnfr.cache import TNFRHierarchicalCache
    >>> my_cache = TNFRHierarchicalCache(max_memory_mb=256)
    >>> @cache_tnfr_computation(
    ...     level=CacheLevel.NODE_PROPERTIES,
    ...     dependencies={'node_data'},
    ...     cache_instance=my_cache
    ... )
    ... def get_node_property(graph, node_id):
    ...     return graph.nodes[node_id]
    """

    def decorator(func: F) -> F:
        func_name = f"{func.__module__}.{func.__qualname__}"
        # A private dependency targets this wrapper alone, including functions
        # with no structural dependencies and separately constructed closures.
        function_dependency = f"__tnfr_function__:{func_name}:{id(func)}"
        structural_dependencies = set(dependencies)
        entry_dependencies = structural_dependencies | {function_dependency}

        @wraps(func)
        def wrapper(*args: Any, **kwargs: Any) -> Any:
            # Get cache instance
            cache = cache_instance if cache_instance is not None else get_global_cache()

            # Generate cache key
            # Bind defaults so implicit kwargs (alpha, landmark_ratio)
            # are included in the cache key deterministically.
            try:
                import inspect

                sig = inspect.signature(func)
                bound = sig.bind_partial(*args, **kwargs)
                bound.apply_defaults()
                norm_kwargs = dict(bound.arguments)
                cache_key = _generate_cache_key(
                    function_dependency,
                    tuple(),
                    norm_kwargs,
                )
            except Exception:
                # Fallback to raw args/kwargs if binding fails
                cache_key = _generate_cache_key(function_dependency, args, kwargs)

            # Append dependency hash to key to detect in-place updates
            graph_obj = _extract_graph_from_args(args, kwargs)
            dep_hash = _compute_dependency_hash(graph_obj, structural_dependencies)
            if dep_hash:
                cache_key = f"{cache_key}:{dep_hash}"

            # Try to get from cache
            cached_result = cache.get(cache_key, level)
            if cached_result is not None:
                return cached_result

            # Compute result
            result = func(*args, **kwargs)

            # Estimate computational cost
            comp_cost = 1.0
            if cost_estimator is not None:
                try:
                    comp_cost = float(cost_estimator(*args, **kwargs))
                except (TypeError, ValueError):
                    comp_cost = 1.0

            # Store in cache
            cache.set(cache_key, result, level, entry_dependencies, comp_cost)

            return result

        # Attach metadata for introspection
        wrapper._cache_level = level  # type: ignore
        wrapper._cache_dependencies = structural_dependencies  # type: ignore
        wrapper._cache_instance = cache_instance  # type: ignore
        wrapper._cache_function_dependency = function_dependency  # type: ignore
        wrapper._is_cached = True  # type: ignore

        return wrapper  # type: ignore

    return decorator


def invalidate_function_cache(func: Callable[..., Any]) -> int:
    """Invalidate cache entries for a specific decorated function.

    Parameters
    ----------
    func : callable
        The decorated function whose cache entries should be invalidated.

    Returns
    -------
    int
        Number of entries invalidated.

    Raises
    ------
    ValueError
        If the function is not decorated with @cache_tnfr_computation.
    """
    if not hasattr(func, "_is_cached"):
        raise TNFRValueError(
            f"Function {func.__name__} is not cached",
            context={"function": func.__name__},
            suggestion="Ensure the function is decorated with @cache_tnfr_computation.",
        )

    cache = getattr(func, "_cache_instance", None)
    if cache is None:
        cache = get_global_cache()
    dependency = getattr(func, "_cache_function_dependency", None)
    return cache.invalidate_by_dependency(dependency) if dependency is not None else 0


# ============================================================================
# Graph Change Tracking (moved from caching/invalidation.py for consolidation)
# ============================================================================


class GraphChangeTracker:
    """Track graph modifications for selective cache invalidation.

    Installs hooks into graph modification methods to automatically invalidate
    affected cache entries when structural properties change.

    Parameters
    ----------
    cache : TNFRHierarchicalCache
        The cache instance to invalidate.

    Attributes
    ----------
    topology_changes : int
        Count of topology modifications (add/remove node/edge).
    property_changes : int
        Count of node property modifications.

    Examples
    --------
    >>> import networkx as nx
    >>> from tnfr.cache import (
    ...     TNFRHierarchicalCache,
    ...     GraphChangeTracker,
    ...     CacheLevel,
    ... )
    >>> cache = TNFRHierarchicalCache()
    >>> G = nx.Graph()
    >>> tracker = GraphChangeTracker(cache)
    >>> tracker.track_graph_changes(G)
    >>> # Now modifications to G will trigger cache invalidation
    >>> cache.set("key1", 1, CacheLevel.GRAPH_STRUCTURE, {'graph_topology'})
    >>> G.add_node("n1")  # Invalidates graph_topology cache entries
    >>> cache.get("key1", CacheLevel.GRAPH_STRUCTURE)  # Returns None
    """

    def __init__(self, cache: TNFRHierarchicalCache):
        self._cache = cache
        self.topology_changes = 0
        self.property_changes = 0
        self._tracked_graphs: weakref.WeakSet[Any] = weakref.WeakSet()

    def track_graph_changes(self, graph: Any) -> None:
        """Install hooks to track changes in a graph.

        Wraps single and bulk node/edge mutations, preserving their arguments
        and return values. Nested calls count as one public operation.

        Parameters
        ----------
        graph : GraphLike
            The graph to monitor for changes.

        Notes
        -----
        This uses monkey-patching to intercept graph modifications. The
        original methods are preserved. Invalidation also follows an exception
        because a bulk operation may have modified a prefix before failing.
        """
        if graph in self._tracked_graphs:
            return  # Already tracking this graph
        self._tracked_graphs.add(graph)
        depth = 0

        def wrap_mutation(original: Callable[..., Any]) -> Callable[..., Any]:
            @wraps(original)
            def tracked(*args: Any, **kwargs: Any) -> Any:
                nonlocal depth
                depth += 1
                try:
                    return original(*args, **kwargs)
                finally:
                    depth -= 1
                    if depth == 0:
                        self._on_topology_change()
                        # clear() removes graph metadata, but not these hooks.
                        graph.graph["_tnfr_change_tracker"] = self

            return tracked

        for name in (
            "add_node",
            "add_nodes_from",
            "remove_node",
            "remove_nodes_from",
            "add_edge",
            "add_edges_from",
            "add_weighted_edges_from",
            "remove_edge",
            "remove_edges_from",
            "clear",
            "clear_edges",
            "update",
        ):
            original = getattr(graph, name, None)
            if callable(original):
                setattr(graph, name, wrap_mutation(original))

        # Store reference to tracker for property changes
        if hasattr(graph, "graph"):
            graph.graph["_tnfr_change_tracker"] = self

    def on_node_property_change(
        self,
        node_id: Any,
        property_name: str,
        old_value: Any | None = None,
        new_value: Any | None = None,
    ) -> None:
        """Notify tracker of a node property change.

        Parameters
        ----------
        node_id : Any
            The node whose property changed.
        property_name : str
            Name of the property that changed (e.g., 'epi', 'vf', 'phase').
        old_value : Any, optional
            Previous value (for logging/debugging).
        new_value : Any, optional
            New value (for logging/debugging).

        Notes
        -----
        This should be called explicitly when node properties are modified
        outside of the graph's standard API (e.g., G.nodes[n]['epi'] = value).
        """
        # Invalidate node-specific dependency
        dep_key = f"node_{property_name}_{node_id}"
        self._cache.invalidate_by_dependency(dep_key)

        # Invalidate global property dependency
        global_dep = f"all_node_{property_name}"
        self._cache.invalidate_by_dependency(global_dep)

        # Match the same canonical aliases used by dependency hashing. Keep
        # legacy fine-grained names above, while invalidating shared readers.
        for dependency in (
            "node_epi",
            "node_vf",
            "node_phase",
            "node_dnfr",
            "node_depi",
        ):
            if (
                property_name in _dependency_alias_keys(dependency)
                or property_name == dependency[5:]
            ):
                canonical_property = dependency[5:]
                for name in (
                    dependency,
                    f"node_{canonical_property}_{node_id}",
                    f"all_node_{canonical_property}",
                ):
                    self._cache.invalidate_by_dependency(name)
        self._cache.invalidate_by_dependency("node_data")

        # Invalidate derived metrics for this node
        if property_name in ["epi", "vf", "phase", "delta_nfr"]:
            self._cache.invalidate_by_dependency(f"derived_metrics_{node_id}")

        self.property_changes += 1

    def _on_topology_change(self) -> None:
        """Handle topology modifications (add/remove node/edge)."""
        # Invalidate topology-dependent caches
        self._cache.invalidate_by_dependency("graph_topology")
        self._cache.invalidate_by_dependency("node_neighbors")
        self._cache.invalidate_by_dependency("adjacency_matrix")

        self.topology_changes += 1

    def reset_counters(self) -> None:
        """Reset change counters."""
        self.topology_changes = 0
        self.property_changes = 0


def track_node_property_update(
    graph: Any,
    node_id: Any,
    property_name: str,
    new_value: Any,
) -> None:
    """Helper to track node property updates.

    Updates the node property and notifies the change tracker if one is
    attached to the graph.

    Parameters
    ----------
    graph : GraphLike
        The graph containing the node.
    node_id : Any
        The node to update.
    property_name : str
        Property name to update.
    new_value : Any
        New value for the property.

    Examples
    --------
    >>> import networkx as nx
    >>> from tnfr.cache import TNFRHierarchicalCache, GraphChangeTracker
    >>> from tnfr.cache import track_node_property_update
    >>> cache = TNFRHierarchicalCache()
    >>> G = nx.Graph()
    >>> G.add_node("n1", epi=0.5)
    >>> tracker = GraphChangeTracker(cache)
    >>> tracker.track_graph_changes(G)
    >>> # Use helper to update and invalidate
    >>> track_node_property_update(G, "n1", "epi", 0.7)
    """
    # Get old value
    old_value = graph.nodes[node_id].get(property_name)

    # Update property
    graph.nodes[node_id][property_name] = new_value

    # Notify tracker if present
    if hasattr(graph, "graph"):
        tracker = graph.graph.get("_tnfr_change_tracker")
        if isinstance(tracker, GraphChangeTracker):
            tracker.on_node_property_change(
                node_id,
                property_name,
                old_value,
                new_value,
            )


# ============================================================================
# Persistent Cache (moved from caching/persistence.py for consolidation)
# ============================================================================


class PersistentTNFRCache:
    """Cache with optional disk persistence for costly computations.

    Combines in-memory caching with selective disk persistence for
    specific cache levels. Expensive computations can be preserved
    between sessions while temporary computations remain memory-only.

    Parameters
    ----------
    cache_dir : Path or str, default: ".tnfr_cache"
        Directory for persistent cache files.
    max_memory_mb : int, default: 512
        Memory limit for in-memory cache.
    persist_levels : set[CacheLevel], optional
        Cache levels to persist to disk. Defaults to GRAPH_STRUCTURE
        and DERIVED_METRICS.

    Examples
    --------
    >>> from pathlib import Path
    >>> from tnfr.cache import PersistentTNFRCache, CacheLevel
    >>> cache = PersistentTNFRCache(cache_dir=Path("/tmp/tnfr_cache"))
    >>> # Cache is automatically persisted for expensive operations
    >>> cache.set_persistent(
    ...     "coherence_large_graph",
    ...     0.95,
    ...     CacheLevel.DERIVED_METRICS,
    ...     dependencies={'graph_topology'},
    ...     computation_cost=1000.0,
    ...     persist_to_disk=True
    ... )
    >>> # Later, in a new session
    >>> result = cache.get_persistent(
    ...     "coherence_large_graph",
    ...     CacheLevel.DERIVED_METRICS,
    ... )
    """

    def __init__(
        self,
        cache_dir: Any = ".tnfr_cache",  # Path | str
        max_memory_mb: int = 512,
        persist_levels: set[CacheLevel] | None = None,
    ):
        from pathlib import Path

        self.cache_dir = Path(cache_dir)
        self.cache_dir.mkdir(exist_ok=True, parents=True)
        self._memory_cache = TNFRHierarchicalCache(max_memory_mb=max_memory_mb)

        if persist_levels is None:
            persist_levels = {
                CacheLevel.GRAPH_STRUCTURE,
                CacheLevel.DERIVED_METRICS,
            }
        self._persist_levels = persist_levels

    def get_persistent(self, key: str, level: CacheLevel) -> Any | None:
        """Retrieve value from memory cache, falling back to disk.

        Parameters
        ----------
        key : str
            Cache key.
        level : CacheLevel
            Cache level.

        Returns
        -------
        Any or None
            Cached value if found, None otherwise.
        """
        # Try memory first
        result = self._memory_cache.get(key, level)
        if result is not None:
            return result

        # Try disk if level is persisted
        if level in self._persist_levels:
            file_path = self._get_cache_file_path(key, level)
            if file_path.exists():
                try:
                    with open(file_path, "rb") as f:
                        cached_data = pickle.load(f)

                    # Validate structure
                    if not isinstance(cached_data, dict):
                        file_path.unlink(missing_ok=True)
                        return None

                    value = cached_data.get("value")
                    dependencies = cached_data.get("dependencies", set())
                    computation_cost = cached_data.get("computation_cost", 1.0)

                    # Load back into memory cache
                    self._memory_cache.set(
                        key,
                        value,
                        level,
                        dependencies,
                        computation_cost,
                    )

                    return value

                except (pickle.PickleError, EOFError, OSError):
                    # Corrupt cache file, remove it
                    file_path.unlink(missing_ok=True)

        return None

    def set_persistent(
        self,
        key: str,
        value: Any,
        level: CacheLevel,
        dependencies: set[str],
        computation_cost: float = 1.0,
        persist_to_disk: bool = True,
    ) -> None:
        """Store value in memory and optionally persist to disk.

        Parameters
        ----------
        key : str
            Cache key.
        value : Any
            Value to cache.
        level : CacheLevel
            Cache level.
        dependencies : set[str]
            Structural dependencies.
        computation_cost : float, default: 1.0
            Computation cost estimate.
        persist_to_disk : bool, default: True
            Whether to persist this entry to disk.
        """
        # Always store in memory
        self._memory_cache.set(
            key,
            value,
            level,
            dependencies,
            computation_cost,
        )

        # Persist to disk if requested and level supports it
        if persist_to_disk and level in self._persist_levels:
            file_path = self._get_cache_file_path(key, level)
            cache_data = {
                "value": value,
                "dependencies": dependencies,
                "computation_cost": computation_cost,
                "timestamp": time.time(),
            }

            try:
                with open(file_path, "wb") as f:
                    pickle.dump(cache_data, f, protocol=pickle.HIGHEST_PROTOCOL)
            except (pickle.PickleError, OSError):
                # A failed write must not leave a partial cache to be loaded.
                file_path.unlink(missing_ok=True)
        else:
            # A memory-only replacement must not revive an older disk value
            # after memory eviction or a process restart.
            (self.cache_dir / level.value / f"{key}.pkl").unlink(missing_ok=True)

    def invalidate_by_dependency(self, dependency: str) -> int:
        """Invalidate dependent memory entries and discard disk snapshots.

        The legacy disk format has no authenticated dependency index. Disk
        invalidation is conservative: all snapshots are discarded, without
        unpickling them to inspect dependencies. Unaffected memory survives.

        Parameters
        ----------
        dependency : str
            The structural property that changed.

        Returns
        -------
        int
            Number of entries invalidated from memory.
        """
        # Invalidate memory cache
        count = self._memory_cache.invalidate_by_dependency(dependency)

        self.clear_persistent_cache()
        return count

    def clear_persistent_cache(self, level: CacheLevel | None = None) -> None:
        """Clear persistent cache files.

        Parameters
        ----------
        level : CacheLevel, optional
            Specific level to clear. If None, clears all levels.
        """
        if level is not None:
            level_dir = self.cache_dir / level.value
            if level_dir.exists():
                for file_path in level_dir.glob("*.pkl"):
                    file_path.unlink(missing_ok=True)
        else:
            # Clear all levels
            for file_path in self.cache_dir.rglob("*.pkl"):
                file_path.unlink(missing_ok=True)

    def cleanup_old_entries(self, max_age_days: int = 30) -> int:
        """Remove old cache files from disk.

        Parameters
        ----------
        max_age_days : int, default: 30
            Maximum age in days before removal.

        Returns
        -------
        int
            Number of files removed.
        """
        count = 0
        max_age_seconds = max_age_days * 24 * 3600
        current_time = time.time()

        for file_path in self.cache_dir.rglob("*.pkl"):
            try:
                mtime = file_path.stat().st_mtime
                if current_time - mtime > max_age_seconds:
                    file_path.unlink()
                    count += 1
            except OSError:
                continue

        return count

    def get_stats(self) -> dict[str, Any]:
        """Get combined statistics from memory and disk cache.

        Returns
        -------
        dict[str, Any]
            Statistics including memory stats and disk usage.
        """
        stats = self._memory_cache.get_stats()

        # Add disk stats
        disk_files = 0
        disk_size_bytes = 0
        for file_path in self.cache_dir.rglob("*.pkl"):
            disk_files += 1
            try:
                disk_size_bytes += file_path.stat().st_size
            except OSError:
                continue

        stats["disk_files"] = disk_files
        stats["disk_size_mb"] = disk_size_bytes / (1024 * 1024)

        return stats

    def _get_cache_file_path(self, key: str, level: CacheLevel) -> Any:
        """Get file path for a cache entry.

        Organizes cache files by level in subdirectories.
        """
        level_dir = self.cache_dir / level.value
        level_dir.mkdir(exist_ok=True, parents=True)
        # Use key as filename (already hashed in decorator)
        return level_dir / f"{key}.pkl"