# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/when-must-an-instance-defer-semantic-promotion-during-replay-when-episodic-traces-exhibit-cross-session-interference.md
# Remnant Warrant:    https://corpusfetch.com/c/when-must-an-instance-defer-semantic-promotion-during-replay-when-episodic-traces-exhibit-cross-session-interference.warrant.json
# Markdown SHA-256:   f8992f01987ce2e7a248c6ec90d83a0d66d0c9d803e2f5785eda6ba2ef548fb4
# Code AST SHA-256:   3aaa589b2a8fae90696e5172625432baa8c71d1f3fcbda260bb9755a1f5e2902
# Verified Classes:   ContaminatedPromotionError, SessionSupport, Candidate
# Defensive Guards:   1 fail-closed invariants (zero external deps)

import math
from dataclasses import dataclass, field
from typing import Dict, List


class ContaminatedPromotionError(RuntimeError):
    """Raised when a candidate generalization is promoted despite session-concentrated
    support and unannotated recent contradiction."""


@dataclass
class SessionSupport:
    session_id: str
    replayed: int
    supporting: int

    @property
    def density(self) -> float:
        return self.supporting / self.replayed if self.replayed else 0.0


@dataclass
class Candidate:
    rule: str
    per_session: List[SessionSupport] = field(default_factory=list)
    environment_change_logged: bool = False


def evaluate_promotion(
    candidate: Candidate,
    tau_promote: float = 0.75,
    rho_max: float = 2.0,
    min_replayed_per_session: int = 2,
) -> str:
    """Return 'promote', 'defer', or 'reject'. Raise on the contaminated case
    only when the caller forces promotion past a deferral."""
    eligible = [s for s in candidate.per_session if s.replayed >= min_replayed_per_session]
    if not eligible:
        return "reject"

    densities = [s.density for s in eligible]
    overall = sum(s.supporting for s in eligible) / sum(s.replayed for s in eligible)
    if overall < tau_promote:
        return "reject"

    lo, hi = min(densities), max(densities)
    if lo <= 0.0:
        # A session with zero support among eligible sessions: either the world
        # changed (promote, scoped) or the candidate is contaminated (defer).
        return "promote_scoped" if candidate.environment_change_logged else "defer"

    ratio = hi / lo
    if ratio > rho_max and not candidate.environment_change_logged:
        return "defer"
    return "promote"


def force_promote(candidate: Candidate, **kwargs) -> str:
    verdict = evaluate_promotion(candidate, **kwargs)
    if verdict == "defer":
        raise ContaminatedPromotionError(
            f"Refusing to promote {candidate.rule!r}: session support is "
            f"concentrated (ratio > rho_max) with recent contradiction and no "
            f"logged environment change. Oversample the dissenting session and "
            f"re-evaluate next replay cycle."
        )
    return verdict