# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/which-concurrency-lock-contention-patterns-must-be-recorded-before-worker-pools-spin-down.md
# Remnant Warrant:    https://corpusfetch.com/c/which-concurrency-lock-contention-patterns-must-be-recorded-before-worker-pools-spin-down.warrant.json
# Markdown SHA-256:   8fdc5cae48c253f0c6650c7c23c1558ac86c8ca6ad865de2e18c2e0cd7763821
# Code AST SHA-256:   8885294672f012c0ca81161d83ffda4f116b795a10477f90cde0bc2293b245f1
# Verified Classes:   ContentionVerdict, LockAcquisitionEvent, ContentionSnapshot, LockContentionManifest, LockContentionRecorder
# Defensive Guards:   1 fail-closed invariants (zero external deps)

from dataclasses import dataclass
from enum import Enum, auto
import hashlib
import json


class ContentionVerdict(Enum):
    LOW_CONTENTION = auto()
    HOT_LOCK_DETECTED = auto()
    STARVATION_OBSERVED = auto()
    CYCLIC_ORDERING_DETECTED = auto()


@dataclass(frozen=True)
class LockAcquisitionEvent:
    worker_id: str
    lock_id: str
    wait_time_ms: int
    hold_time_ms: int


@dataclass(frozen=True)
class ContentionSnapshot:
    total_pool_wait_ms: int
    hot_lock_threshold_pct: float
    starvation_ratio: float


@dataclass(frozen=True)
class LockContentionManifest:
    verdict: ContentionVerdict
    hot_lock_register: tuple[tuple[str, int], ...]
    starvation_signatures: tuple[tuple[str, str], ...]
    cyclic_ordering_witness: tuple[tuple[str, str], ...]
    recommended_pool_size: int
    manifest_digest: str


class LockContentionRecorder:
    """
    Evaluates worker pool lock acquisitions at spin-down to preserve
    contention topology for safe successor pool reconfiguration.
    """

    @staticmethod
    def _hash(data: str) -> str:
        return hashlib.sha256(data.encode("utf-8")).hexdigest()

    @classmethod
    def evaluate_contention(
        cls,
        events: tuple[LockAcquisitionEvent, ...],
        snapshot: ContentionSnapshot,
        nominal_pool_size: int,
    ) -> LockContentionManifest:
        
        if nominal_pool_size <= 0:
            raise ValueError("Nominal pool size must be positive.")

        # Gate 1: Aggregate wait time per lock
        lock_wait: dict[str, int] = {}
        worker_wait: dict[str, int] = {}

        for event in events:
            lock_wait[event.lock_id] = lock_wait.get(event.lock_id, 0) + event.wait_time_ms
            worker_wait[event.worker_id] = worker_wait.get(event.worker_id, 0) + event.wait_time_ms

        total_wait = sum(lock_wait.values()) or 1  # prevent div by zero

        # Gate 2: Identify hot locks
        hot_locks = {
            lock_id: wait
            for lock_id, wait in lock_wait.items()
            if (wait / total_wait) * 100 >= snapshot.hot_lock_threshold_pct
        }

        # Gate 3: Identify starvation signatures
        starvation: list[tuple[str, str]] = []
        if worker_wait:
            median_wait = sorted(worker_wait.values())[len(worker_wait) // 2]
            if median_wait > 0:
                for worker_id, wait in worker_wait.items():
                    if wait > median_wait * snapshot.starvation_ratio:
                        starvation.append((worker_id, f"wait={wait}ms, median={median_wait}ms"))

        # Gate 4: Determine verdict
        if hot_locks and starvation:
            verdict = ContentionVerdict.STARVATION_OBSERVED
        elif hot_locks:
            verdict = ContentionVerdict.HOT_LOCK_DETECTED
        else:
            verdict = ContentionVerdict.LOW_CONTENTION

        # Recommended pool size: shrink if hot locks + starvation observed
        if verdict is ContentionVerdict.STARVATION_OBSERVED:
            recommended = max(1, nominal_pool_size // 2)
        elif verdict is ContentionVerdict.HOT_LOCK_DETECTED:
            recommended = max(1, int(nominal_pool_size * 0.75))
        else:
            recommended = nominal_pool_size

        digest_payload = {
            "verdict": verdict.name,
            "hot_locks": tuple(sorted(hot_locks.keys())),
            "starvation_count": str(len(starvation)),
            "pool_size": str(recommended),
        }
        
        digest = cls._hash(
            json.dumps(digest_payload, sort_keys=True, separators=(",", ":"))
        )

        return LockContentionManifest(
            verdict=verdict,
            hot_lock_register=tuple(sorted(hot_locks.items())),
            starvation_signatures=tuple(starvation),
            cyclic_ordering_witness=tuple(),
            recommended_pool_size=recommended,
            manifest_digest=digest,
        )

    @classmethod
    def may_spawn_pool_size(
        cls, manifest: LockContentionManifest, requested_size: int
    ) -> tuple[bool, str]:
        """
        Fetching-time check: confirms a successor may spawn the requested pool
        size without re-triggering contention.
        """
        if requested_size > manifest.recommended_pool_size:
            return False, (
                f"POOL_SIZE_VIOLATION: Requested {requested_size} exceeds "
                f"recommended {manifest.recommended_pool_size} based on "
                f"verdict {manifest.verdict.name}."
            )
        return True, "POOL_SIZE_ADMISSIBLE"