metrics/stability.py

"""
Generic stability metrics for WorldSmithAI.

This module computes stability over arbitrary world collections. It deliberately
avoids hardcoded species, farm, research, market, transport, or civilization
assumptions.

A stability metric can measure:
    - population stability by agent type
    - resource amount stability by resource type
    - goal distribution stability
    - market-role stability
    - infrastructure-status stability
    - strategy/adoption stability
    - any DSL-defined grouped numeric path

Example:
    metric = StabilityMetric(collection="agents", group_by_path="type")
    result = metric.compute(world)

    print(result.variance)
    print(result.stability_score)

Temporal tracking:
    tracker = StabilityTracker(metric)

    for _ in range(100):
        world.step()
        tracker.update(world)

Future extensibility:
    - Add rolling-window variance and volatility.
    - Add autocorrelation and trend stability.
    - Add equilibrium detection.
    - Add collapse and boom detection.
    - Add stability contributions per group.
    - Feed stability signals into interestingness and God-Agent metrics.
"""

from __future__ import annotations

import copy
import logging
import math
from collections.abc import Iterable, Mapping, MutableSequence, Sequence
from dataclasses import dataclass, field
from enum import Enum
from numbers import Real
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, ClassVar

import numpy as np

if TYPE_CHECKING:
    from core.world import World


logger = logging.getLogger(__name__)

_MISSING = object()
_EPSILON = 1.0e-12


class StabilityCollection(str, Enum):
    """Collections over which stability may be computed."""

    AGENTS = "agents"
    RESOURCES = "resources"
    BOTH = "both"


class StabilityAggregation(str, Enum):
    """Aggregation modes used to convert world objects into group values."""

    COUNT = "count"
    SUM = "sum"
    MEAN = "mean"


@dataclass(frozen=True)
class StabilityResult:
    """Result produced by stability metric computation.

    Attributes:
        metric_name: Stable metric name.
        collection: Collection analyzed.
        group_by_path: Path used to group objects.
        value_path: Optional numeric path used for sum or mean aggregation.
        weight_path: Optional numeric path used as contribution weight.
        aggregation: Aggregation mode.
        item_count: Number of included objects.
        group_count: Number of groups after aggregation.
        group_values: Current value per group.
        previous_group_values: Previous value per group when supplied.
        deltas: Current minus previous value per group when supplied.
        total_value: Sum of current group values.
        mean_value: Mean group value.
        variance: Population variance across current group values.
        standard_deviation: Population standard deviation across group values.
        coefficient_of_variation: Standard deviation divided by absolute mean.
        min_value: Minimum group value.
        max_value: Maximum group value.
        value_range: Difference between max and min.
        normalized_range: Range divided by absolute mean.
        temporal_variance: Variance of group deltas when previous values exist.
        temporal_standard_deviation: Standard deviation of deltas.
        total_absolute_delta: Sum of absolute group deltas.
        mean_absolute_delta: Mean absolute group delta.
        mean_relative_delta: Mean relative group delta.
        stability_score: Bounded score in [0, 1], where higher is more stable.
        step: Optional world step.
        metadata: Additional JSON-compatible details.
    """

    metric_name: str
    collection: str
    group_by_path: str
    value_path: str | None
    weight_path: str | None
    aggregation: str
    item_count: int
    group_count: int
    group_values: Mapping[str, float] = field(default_factory=dict)
    previous_group_values: Mapping[str, float] = field(default_factory=dict)
    deltas: Mapping[str, float] = field(default_factory=dict)
    total_value: float = 0.0
    mean_value: float = 0.0
    variance: float = 0.0
    standard_deviation: float = 0.0
    coefficient_of_variation: float = 0.0
    min_value: float = 0.0
    max_value: float = 0.0
    value_range: float = 0.0
    normalized_range: float = 0.0
    temporal_variance: float = 0.0
    temporal_standard_deviation: float = 0.0
    total_absolute_delta: float = 0.0
    mean_absolute_delta: float = 0.0
    mean_relative_delta: float = 0.0
    stability_score: float = 1.0
    step: int | None = None
    metadata: Mapping[str, Any] = field(default_factory=dict)

    @property
    def is_empty(self) -> bool:
        """Return whether no objects contributed to this result."""

        return self.item_count == 0 or self.group_count == 0

    @property
    def is_temporal(self) -> bool:
        """Return whether this result includes previous-state comparison."""

        return bool(self.previous_group_values)

    def to_dict(self) -> dict[str, Any]:
        """Return a JSON-friendly representation of the stability result."""

        return {
            "metric_name": self.metric_name,
            "collection": self.collection,
            "group_by_path": self.group_by_path,
            "value_path": self.value_path,
            "weight_path": self.weight_path,
            "aggregation": self.aggregation,
            "item_count": self.item_count,
            "group_count": self.group_count,
            "group_values": copy.deepcopy(dict(self.group_values)),
            "previous_group_values": copy.deepcopy(dict(self.previous_group_values)),
            "deltas": copy.deepcopy(dict(self.deltas)),
            "total_value": self.total_value,
            "mean_value": self.mean_value,
            "variance": self.variance,
            "standard_deviation": self.standard_deviation,
            "coefficient_of_variation": self.coefficient_of_variation,
            "min_value": self.min_value,
            "max_value": self.max_value,
            "value_range": self.value_range,
            "normalized_range": self.normalized_range,
            "temporal_variance": self.temporal_variance,
            "temporal_standard_deviation": self.temporal_standard_deviation,
            "total_absolute_delta": self.total_absolute_delta,
            "mean_absolute_delta": self.mean_absolute_delta,
            "mean_relative_delta": self.mean_relative_delta,
            "stability_score": self.stability_score,
            "is_empty": self.is_empty,
            "is_temporal": self.is_temporal,
            "step": self.step,
            "metadata": copy.deepcopy(dict(self.metadata)),
        }


@dataclass
class StabilityMetric:
    """Compute generic stability over grouped world objects.

    The default configuration computes population stability over alive agents
    grouped by ``type``. This can represent species, roles, professions,
    factions, node classes, strategies, or any DSL-defined type label.

    Examples:
        Agent population stability:
            metric = StabilityMetric()
            result = metric.compute(world)

        Resource amount stability:
            metric = StabilityMetric(
                collection="resources",
                group_by_path="type",
                value_path="amount",
                aggregation="sum",
                alive_only=False,
            )
            result = metric.compute(world)

        Goal distribution stability:
            metric = StabilityMetric(
                collection="agents",
                group_by_path="state.current_goal",
                include_missing=True,
            )
            result = metric.compute(world)
    """

    name: ClassVar[str] = "stability"

    collection: StabilityCollection | str = StabilityCollection.AGENTS
    group_by_path: str = "type"
    value_path: str | None = None
    weight_path: str | None = None
    aggregation: StabilityAggregation | str = StabilityAggregation.COUNT

    alive_only: bool = True
    include_missing: bool = False
    missing_label: str = "unknown"

    include_group_values: tuple[str, ...] = ()
    exclude_group_values: tuple[str, ...] = ()

    case_sensitive: bool = True
    strip_labels: bool = True
    include_collection_prefix: bool = False

    minimum_weight: float = 0.0
    metadata: Mapping[str, Any] = field(default_factory=dict)

    def compute(
        self,
        world: World,
        *,
        previous_values: Mapping[str, float] | None = None,
    ) -> StabilityResult:
        """Compute stability for the configured world collection.

        Args:
            world: Runtime world object.
            previous_values: Optional previous group values. When supplied,
                temporal stability statistics are included.

        Returns:
            ``StabilityResult`` with snapshot and optional temporal statistics.
        """

        collection = _normalize_collection(self.collection)
        aggregation = _normalize_aggregation(self.aggregation)
        items = _iter_world_items(world, collection)
        group_values, included_count, skipped_count = self._group_values(items, collection, aggregation)

        result = stability_from_values(
            group_values,
            previous_values=previous_values,
            metric_name=self.name,
            collection=collection.value,
            group_by_path=self.group_by_path,
            value_path=self.value_path,
            weight_path=self.weight_path,
            aggregation=aggregation.value,
            item_count=included_count,
            step=_world_step(world),
            metadata={
                **copy.deepcopy(dict(self.metadata)),
                "skipped_count": skipped_count,
                "alive_only": self.alive_only,
                "include_missing": self.include_missing,
                "minimum_weight": self.minimum_weight,
            },
        )

        logger.debug(
            "Computed stability over %s grouped by %s: variance=%.6f score=%.6f",
            collection.value,
            self.group_by_path,
            result.variance,
            result.stability_score,
        )

        return result

    def __call__(
        self,
        world: World,
        *,
        previous_values: Mapping[str, float] | None = None,
    ) -> StabilityResult:
        """Compute stability, allowing metric instances to be called directly."""

        return self.compute(world, previous_values=previous_values)

    def _group_values(
        self,
        items: Sequence[Any],
        collection: StabilityCollection,
        aggregation: StabilityAggregation,
    ) -> tuple[dict[str, float], int, int]:
        """Aggregate world objects into group values."""

        group_sums: dict[str, float] = {}
        group_weights: dict[str, float] = {}
        included_count = 0
        skipped_count = 0

        include_values = {
            _normalize_label(value, case_sensitive=self.case_sensitive, strip=True)
            for value in self.include_group_values
        }
        exclude_values = {
            _normalize_label(value, case_sensitive=self.case_sensitive, strip=True)
            for value in self.exclude_group_values
        }

        for item in items:
            if self.alive_only and _is_agent_like(item) and not _is_alive(item):
                skipped_count += 1
                continue

            label = self._label_for(item, collection)
            if label is None:
                skipped_count += 1
                continue

            normalized_label = _normalize_label(
                label,
                case_sensitive=self.case_sensitive,
                strip=self.strip_labels,
            )

            if include_values and normalized_label not in include_values:
                skipped_count += 1
                continue

            if exclude_values and normalized_label in exclude_values:
                skipped_count += 1
                continue

            weight = self._weight_for(item)
            if weight <= max(0.0, float(self.minimum_weight)):
                skipped_count += 1
                continue

            value = self._value_for(item, aggregation)
            if value is None:
                skipped_count += 1
                continue

            if aggregation is StabilityAggregation.COUNT:
                contribution = weight
                denominator = 1.0
            elif aggregation is StabilityAggregation.SUM:
                contribution = value * weight
                denominator = 1.0
            else:
                contribution = value * weight
                denominator = weight

            group_sums[normalized_label] = group_sums.get(normalized_label, 0.0) + contribution
            group_weights[normalized_label] = group_weights.get(normalized_label, 0.0) + denominator
            included_count += 1

        if aggregation is StabilityAggregation.MEAN:
            group_values = {
                label: group_sums[label] / max(group_weights.get(label, 0.0), _EPSILON)
                for label in group_sums
            }
        else:
            group_values = group_sums

        return dict(sorted(group_values.items(), key=lambda item: item[0])), included_count, skipped_count

    def _label_for(self, item: Any, collection: StabilityCollection) -> str | None:
        """Return the group label for an item."""

        raw_label = _read_path(item, self.group_by_path, _MISSING)

        if raw_label is _MISSING or raw_label is None or str(raw_label) == "":
            if not self.include_missing:
                return None
            raw_label = self.missing_label

        label = _stable_label(raw_label)

        if self.include_collection_prefix and collection is StabilityCollection.BOTH:
            return f"{_item_collection_name(item)}:{label}"

        return label

    def _weight_for(self, item: Any) -> float:
        """Return the non-negative item weight."""

        if self.weight_path is None:
            return 1.0

        raw_weight = _read_path(item, self.weight_path, 0.0)
        if not _is_number(raw_weight):
            return 0.0

        return max(0.0, float(raw_weight))

    def _value_for(self, item: Any, aggregation: StabilityAggregation) -> float | None:
        """Return the numeric value contributed by an item."""

        if aggregation is StabilityAggregation.COUNT:
            return 1.0

        if self.value_path is None:
            return 1.0

        raw_value = _read_path(item, self.value_path, _MISSING)
        if not _is_number(raw_value):
            return None

        return float(raw_value)


@dataclass
class StabilityTracker:
    """Track stability over time.

    ``StabilityMetric`` can compute one snapshot. ``StabilityTracker`` stores
    previous group values and produces temporal delta statistics at each update.
    """

    metric: StabilityMetric = field(default_factory=StabilityMetric)
    max_history: int = 1000
    history: list[StabilityResult] = field(default_factory=list)
    previous_values: Mapping[str, float] = field(default_factory=dict)

    def update(self, world: World) -> StabilityResult:
        """Compute stability and append the result to history."""

        result = self.metric.compute(world, previous_values=self.previous_values)
        self.previous_values = copy.deepcopy(dict(result.group_values))
        _append_bounded(self.history, result, self.max_history)
        return result

    def latest(self) -> StabilityResult | None:
        """Return the latest stability result, if any."""

        if not self.history:
            return None
        return self.history[-1]

    def reset(self) -> None:
        """Clear tracked history and previous values."""

        self.history.clear()
        self.previous_values = {}

    def to_series(self, value_key: str = "stability_score") -> list[dict[str, Any]]:
        """Return a JSON-friendly time series for one result field.

        Args:
            value_key: Name of a ``StabilityResult`` attribute to extract.

        Returns:
            List of dictionaries with ``index``, ``step``, and ``value``.
        """

        series: list[dict[str, Any]] = []

        for index, result in enumerate(self.history):
            value = getattr(result, value_key, None)
            if not _is_number(value):
                continue

            series.append(
                {
                    "index": index,
                    "step": result.step,
                    "value": float(value),
                }
            )

        return series

    def to_dict(self) -> dict[str, Any]:
        """Return a JSON-friendly tracker representation."""

        return {
            "metric": {
                "name": self.metric.name,
                "collection": _normalize_collection(self.metric.collection).value,
                "group_by_path": self.metric.group_by_path,
                "value_path": self.metric.value_path,
                "weight_path": self.metric.weight_path,
                "aggregation": _normalize_aggregation(self.metric.aggregation).value,
            },
            "max_history": self.max_history,
            "previous_values": copy.deepcopy(dict(self.previous_values)),
            "history": [result.to_dict() for result in self.history],
        }


def compute_stability(
    world: World,
    *,
    collection: StabilityCollection | str = StabilityCollection.AGENTS,
    group_by_path: str = "type",
    value_path: str | None = None,
    weight_path: str | None = None,
    aggregation: StabilityAggregation | str = StabilityAggregation.COUNT,
    alive_only: bool = True,
    include_missing: bool = False,
    previous_values: Mapping[str, float] | None = None,
) -> StabilityResult:
    """Convenience function for computing generic stability."""

    metric = StabilityMetric(
        collection=collection,
        group_by_path=group_by_path,
        value_path=value_path,
        weight_path=weight_path,
        aggregation=aggregation,
        alive_only=alive_only,
        include_missing=include_missing,
    )
    return metric.compute(world, previous_values=previous_values)


def compute_agent_population_stability(
    world: World,
    *,
    alive_only: bool = True,
    include_missing: bool = False,
    previous_values: Mapping[str, float] | None = None,
) -> StabilityResult:
    """Compute population stability over agent ``type`` labels."""

    return compute_stability(
        world,
        collection=StabilityCollection.AGENTS,
        group_by_path="type",
        aggregation=StabilityAggregation.COUNT,
        alive_only=alive_only,
        include_missing=include_missing,
        previous_values=previous_values,
    )


def compute_resource_amount_stability(
    world: World,
    *,
    include_missing: bool = False,
    previous_values: Mapping[str, float] | None = None,
) -> StabilityResult:
    """Compute resource amount stability over resource ``type`` labels."""

    return compute_stability(
        world,
        collection=StabilityCollection.RESOURCES,
        group_by_path="type",
        value_path="amount",
        aggregation=StabilityAggregation.SUM,
        alive_only=False,
        include_missing=include_missing,
        previous_values=previous_values,
    )


def stability_from_values(
    group_values: Mapping[str, float],
    *,
    previous_values: Mapping[str, float] | None = None,
    metric_name: str = "stability",
    collection: str = "custom",
    group_by_path: str = "group",
    value_path: str | None = None,
    weight_path: str | None = None,
    aggregation: str = "count",
    item_count: int | None = None,
    step: int | None = None,
    metadata: Mapping[str, Any] | None = None,
) -> StabilityResult:
    """Compute stability statistics from current and optional previous values.

    Args:
        group_values: Current group values.
        previous_values: Optional previous group values.
        metric_name: Name stored in the result.
        collection: Collection label stored in the result.
        group_by_path: Grouping path stored in the result.
        value_path: Optional numeric value path stored in the result.
        weight_path: Optional weight path stored in the result.
        aggregation: Aggregation label stored in the result.
        item_count: Optional number of contributing items.
        step: Optional world step.
        metadata: Optional result metadata.

    Returns:
        ``StabilityResult`` with cross-sectional and temporal statistics.
    """

    cleaned_values = {
        str(label): float(value)
        for label, value in group_values.items()
        if _is_number(value) and math.isfinite(float(value))
    }
    sorted_values = dict(sorted(cleaned_values.items(), key=lambda item: item[0]))

    if not sorted_values:
        return StabilityResult(
            metric_name=metric_name,
            collection=collection,
            group_by_path=group_by_path,
            value_path=value_path,
            weight_path=weight_path,
            aggregation=aggregation,
            item_count=0 if item_count is None else int(item_count),
            group_count=0,
            step=step,
            metadata=metadata or {},
        )

    values_array = np.asarray(list(sorted_values.values()), dtype=float)
    total_value = float(np.sum(values_array))
    mean_value = float(np.mean(values_array))
    variance = float(np.var(values_array))
    standard_deviation = float(np.sqrt(variance))
    coefficient_of_variation = (
        float(standard_deviation / abs(mean_value))
        if abs(mean_value) > _EPSILON
        else 0.0
    )
    min_value = float(np.min(values_array))
    max_value = float(np.max(values_array))
    value_range = max_value - min_value
    normalized_range = (
        float(value_range / abs(mean_value))
        if abs(mean_value) > _EPSILON
        else 0.0
    )

    previous_cleaned = _clean_previous_values(previous_values)
    deltas = _compute_deltas(sorted_values, previous_cleaned)
    temporal_statistics = _temporal_statistics(deltas, previous_cleaned)

    stability_score = _stability_score(
        coefficient_of_variation=coefficient_of_variation,
        normalized_range=normalized_range,
        mean_relative_delta=temporal_statistics["mean_relative_delta"],
    )

    return StabilityResult(
        metric_name=metric_name,
        collection=collection,
        group_by_path=group_by_path,
        value_path=value_path,
        weight_path=weight_path,
        aggregation=aggregation,
        item_count=len(sorted_values) if item_count is None else int(item_count),
        group_count=len(sorted_values),
        group_values=sorted_values,
        previous_group_values=previous_cleaned,
        deltas=deltas,
        total_value=total_value,
        mean_value=mean_value,
        variance=variance,
        standard_deviation=standard_deviation,
        coefficient_of_variation=coefficient_of_variation,
        min_value=min_value,
        max_value=max_value,
        value_range=value_range,
        normalized_range=normalized_range,
        temporal_variance=temporal_statistics["temporal_variance"],
        temporal_standard_deviation=temporal_statistics["temporal_standard_deviation"],
        total_absolute_delta=temporal_statistics["total_absolute_delta"],
        mean_absolute_delta=temporal_statistics["mean_absolute_delta"],
        mean_relative_delta=temporal_statistics["mean_relative_delta"],
        stability_score=stability_score,
        step=step,
        metadata=metadata or {},
    )


def population_variance(values: Sequence[float]) -> float:
    """Return population variance for a numeric sequence."""

    array = _numeric_array(values)
    if array.size == 0:
        return 0.0
    return float(np.var(array))


def coefficient_of_variation(values: Sequence[float]) -> float:
    """Return coefficient of variation for a numeric sequence."""

    array = _numeric_array(values)
    if array.size == 0:
        return 0.0

    mean_value = float(np.mean(array))
    if abs(mean_value) <= _EPSILON:
        return 0.0

    return float(np.std(array) / abs(mean_value))


def relative_delta(current: float, previous: float) -> float:
    """Return a bounded relative delta magnitude.

    If the previous value is zero and the current value is positive, this
    returns ``1.0``. If both are zero, it returns ``0.0``.
    """

    current_value = float(current)
    previous_value = float(previous)

    if abs(previous_value) <= _EPSILON:
        return 0.0 if abs(current_value) <= _EPSILON else 1.0

    return abs(current_value - previous_value) / abs(previous_value)


def _clean_previous_values(previous_values: Mapping[str, float] | None) -> dict[str, float]:
    """Return cleaned previous values."""

    if previous_values is None:
        return {}

    return {
        str(label): float(value)
        for label, value in previous_values.items()
        if _is_number(value) and math.isfinite(float(value))
    }


def _compute_deltas(
    current_values: Mapping[str, float],
    previous_values: Mapping[str, float],
) -> dict[str, float]:
    """Return current-minus-previous deltas over the union of group labels."""

    if not previous_values:
        return {}

    labels = sorted(set(current_values.keys()) | set(previous_values.keys()))
    return {
        label: float(current_values.get(label, 0.0) - previous_values.get(label, 0.0))
        for label in labels
    }


def _temporal_statistics(
    deltas: Mapping[str, float],
    previous_values: Mapping[str, float],
) -> dict[str, float]:
    """Return temporal statistics from group deltas."""

    if not deltas:
        return {
            "temporal_variance": 0.0,
            "temporal_standard_deviation": 0.0,
            "total_absolute_delta": 0.0,
            "mean_absolute_delta": 0.0,
            "mean_relative_delta": 0.0,
        }

    delta_array = np.asarray(list(deltas.values()), dtype=float)
    absolute_deltas = np.abs(delta_array)
    labels = list(deltas.keys())

    relative_deltas = [
        relative_delta(
            current=previous_values.get(label, 0.0) + deltas[label],
            previous=previous_values.get(label, 0.0),
        )
        for label in labels
    ]

    temporal_variance = float(np.var(delta_array))
    temporal_standard_deviation = float(np.sqrt(temporal_variance))

    return {
        "temporal_variance": temporal_variance,
        "temporal_standard_deviation": temporal_standard_deviation,
        "total_absolute_delta": float(np.sum(absolute_deltas)),
        "mean_absolute_delta": float(np.mean(absolute_deltas)),
        "mean_relative_delta": float(np.mean(np.asarray(relative_deltas, dtype=float)))
        if relative_deltas
        else 0.0,
    }


def _stability_score(
    *,
    coefficient_of_variation: float,
    normalized_range: float,
    mean_relative_delta: float,
) -> float:
    """Return a bounded stability score in [0, 1]."""

    instability_pressure = (
        max(0.0, float(coefficient_of_variation))
        + max(0.0, float(normalized_range))
        + max(0.0, float(mean_relative_delta))
    )
    return float(1.0 / (1.0 + instability_pressure))


def _numeric_array(values: Sequence[float]) -> np.ndarray:
    """Convert a sequence to a finite numeric array."""

    array = np.asarray(values, dtype=float).reshape(-1)
    array = np.nan_to_num(array, nan=0.0, posinf=0.0, neginf=0.0)
    return array


def _normalize_collection(value: StabilityCollection | str) -> StabilityCollection:
    """Normalize a stability collection value."""

    if isinstance(value, StabilityCollection):
        return value

    return StabilityCollection(str(value))


def _normalize_aggregation(value: StabilityAggregation | str) -> StabilityAggregation:
    """Normalize a stability aggregation value."""

    if isinstance(value, StabilityAggregation):
        return value

    return StabilityAggregation(str(value))


def _world_step(world: World) -> int | None:
    """Return the current world step if available."""

    value = getattr(world, "step_count", None)
    if isinstance(value, Real) and not isinstance(value, bool):
        return int(value)
    return None


def _is_number(value: Any) -> bool:
    """Return whether a value is a real numeric scalar, excluding booleans."""

    return isinstance(value, (Real, np.integer, np.floating)) and not isinstance(value, bool)


def _is_alive(item: Any) -> bool:
    """Return whether an item is alive when it exposes an ``alive`` field."""

    return bool(getattr(item, "alive", True))


def _is_agent_like(item: Any) -> bool:
    """Return whether an item looks like a runtime agent."""

    return hasattr(item, "alive") or hasattr(item, "behaviors") or hasattr(item, "policy")


def _item_collection_name(item: Any) -> str:
    """Return a best-effort collection name for an item."""

    if _is_agent_like(item):
        return StabilityCollection.AGENTS.value

    if hasattr(item, "amount"):
        return StabilityCollection.RESOURCES.value

    return "items"


def _iter_world_items(world: World, collection: StabilityCollection) -> tuple[Any, ...]:
    """Return world items for a configured stability collection."""

    if collection is StabilityCollection.AGENTS:
        return _iter_collection(getattr(world, "agents", ()))

    if collection is StabilityCollection.RESOURCES:
        return _iter_collection(getattr(world, "resources", ()))

    agents = _iter_collection(getattr(world, "agents", ()))
    resources = _iter_collection(getattr(world, "resources", ()))
    return agents + resources


def _iter_collection(raw_collection: Any) -> tuple[Any, ...]:
    """Return items from a mapping-backed or sequence-backed collection."""

    if raw_collection is None:
        return ()

    if isinstance(raw_collection, Mapping):
        values = raw_collection.values()
    elif isinstance(raw_collection, Iterable) and not isinstance(raw_collection, (str, bytes)):
        values = raw_collection
    else:
        values = (raw_collection,)

    return tuple(item for item in values if item is not None)


def _split_path(path: str) -> tuple[str, ...]:
    """Split a dot-separated path into components."""

    return tuple(part for part in str(path).split(".") if part)


def _get_mapping_path(container: Mapping[str, Any], path: str, default: Any = _MISSING) -> Any:
    """Read a nested mapping value using dot notation."""

    parts = _split_path(path)
    if not parts:
        return default

    current: Any = container
    for part in parts:
        if not isinstance(current, Mapping) or part not in current:
            return default
        current = current[part]

    return current


def _get_object_path(root: Any, path: str, default: Any = _MISSING) -> Any:
    """Read nested values from mappings, sequences, or object attributes."""

    parts = _split_path(path)
    if not parts:
        return root

    current: Any = root
    for part in parts:
        if isinstance(current, Mapping):
            if part not in current:
                return default
            current = current[part]
            continue

        if isinstance(current, Sequence) and not isinstance(current, (str, bytes)) and part.isdigit():
            index = int(part)
            if index >= len(current):
                return default
            current = current[index]
            continue

        if not hasattr(current, part):
            return default

        current = getattr(current, part)

    return current


def _read_path(item: Any, path: str, default: Any = _MISSING) -> Any:
    """Read a path from an item.

    Supported special prefixes:
        - ``state.foo``
        - ``state:foo``
        - ``memory.foo``
        - ``memory:foo``
        - ``metadata.foo``
        - ``metadata:foo``

    Without a prefix, object attributes and mapping keys are read directly.
    """

    normalized_path = str(path)

    if normalized_path.startswith("state."):
        state = getattr(item, "state", {})
        return _get_mapping_path(
            state if isinstance(state, Mapping) else {},
            normalized_path.removeprefix("state."),
            default,
        )

    if normalized_path.startswith("state:"):
        state = getattr(item, "state", {})
        return _get_mapping_path(
            state if isinstance(state, Mapping) else {},
            normalized_path.removeprefix("state:"),
            default,
        )

    if normalized_path.startswith("memory."):
        memory = getattr(item, "memory", {})
        return _get_mapping_path(
            memory if isinstance(memory, Mapping) else {},
            normalized_path.removeprefix("memory."),
            default,
        )

    if normalized_path.startswith("memory:"):
        memory = getattr(item, "memory", {})
        return _get_mapping_path(
            memory if isinstance(memory, Mapping) else {},
            normalized_path.removeprefix("memory:"),
            default,
        )

    if normalized_path.startswith("metadata."):
        metadata = getattr(item, "metadata", {})
        return _get_mapping_path(
            metadata if isinstance(metadata, Mapping) else {},
            normalized_path.removeprefix("metadata."),
            default,
        )

    if normalized_path.startswith("metadata:"):
        metadata = getattr(item, "metadata", {})
        return _get_mapping_path(
            metadata if isinstance(metadata, Mapping) else {},
            normalized_path.removeprefix("metadata:"),
            default,
        )

    return _get_object_path(item, normalized_path, default)


def _stable_label(value: Any) -> str:
    """Return a stable string label for a group value."""

    if value is None:
        return "none"

    if isinstance(value, bool):
        return "true" if value else "false"

    if _is_number(value):
        numeric_value = float(value)
        if numeric_value.is_integer():
            return str(int(numeric_value))
        return str(numeric_value)

    if isinstance(value, Mapping):
        parts = [f"{key}={_stable_label(value[key])}" for key in sorted(value.keys(), key=str)]
        return "{" + ",".join(parts) + "}"

    if isinstance(value, Sequence) and not isinstance(value, (str, bytes)):
        return "[" + ",".join(_stable_label(item) for item in value) + "]"

    return str(value)


def _normalize_label(value: Any, *, case_sensitive: bool, strip: bool) -> str:
    """Normalize a group label for deterministic comparisons."""

    label = _stable_label(value)
    if strip:
        label = label.strip()
    if not case_sensitive:
        label = label.lower()
    return label


def _append_bounded(items: MutableSequence[Any], value: Any, max_items: int) -> None:
    """Append an item while enforcing an optional maximum history length."""

    items.append(value)
    if max_items > 0 and len(items) > max_items:
        del items[: len(items) - max_items]


METRIC_REGISTRY: Mapping[str, type[StabilityMetric]] = MappingProxyType(
    {
        StabilityMetric.name: StabilityMetric,
    }
)


__all__ = [
    "METRIC_REGISTRY",
    "StabilityAggregation",
    "StabilityCollection",
    "StabilityMetric",
    "StabilityResult",
    "StabilityTracker",
    "coefficient_of_variation",
    "compute_agent_population_stability",
    "compute_resource_amount_stability",
    "compute_stability",
    "population_variance",
    "relative_delta",
    "stability_from_values",
]