# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/when-how-what-must-a-consolidation-scheduler-refuse-to-merge-an-ebbinghaus-decay-bucket-across-a-hierarchical-temporal-i.md
# Remnant Warrant:    https://corpusfetch.com/c/when-how-what-must-a-consolidation-scheduler-refuse-to-merge-an-ebbinghaus-decay-bucket-across-a-hierarchical-temporal-i.warrant.json
# Markdown SHA-256:   9667b53c02adeb5ab54764903e7326d6ca68670ae2d093977918ed3a9054a3a2
# Code AST SHA-256:   fbbed711d8c76724427d455f6ccd9087893d114658101380a6e94a4179842b72
# Verified Classes:   BoundaryMergeRefused, Trace
# Defensive Guards:   5 fail-closed invariants (zero external deps)

import math
from dataclasses import dataclass
from typing import Sequence


class BoundaryMergeRefused(RuntimeError):
    """Raised when a consolidation pass would violate the decay-boundary invariant."""


@dataclass(frozen=True)
class Trace:
    t: float          # timestamp, seconds since epoch
    stability: float  # S, seconds
    rehearsed_since_open: bool = False


def consolidate_bucket(
    traces: Sequence[Trace],
    window_start: float,
    window_end: float,
    seal_watermark: float,
    dispersion_tolerance: float = 0.3,
) -> float:
    """Return the merged stability S for a bucket, or refuse."""
    if not traces:
        raise BoundaryMergeRefused("empty bucket: nothing to consolidate")

    # Invariant 3: half-open interval, watermark must not have passed end.
    if seal_watermark > window_end:
        raise BoundaryMergeRefused(
            f"seal watermark {seal_watermark} exceeds window end {window_end}"
        )

    # Invariant 4: rehearsal mutates S mid-window.
    if any(tr.rehearsed_since_open for tr in traces):
        raise BoundaryMergeRefused(
            "trace rehearsed since window opened; stability mutated mid-window"
        )

    # Invariant 2: dispersion vs. minimum stability (Jensen bias guard).
    t_min = min(tr.t for tr in traces)
    t_max = max(tr.t for tr in traces)
    s_min = min(tr.stability for tr in traces)
    dispersion = t_max - t_min
    if dispersion > dispersion_tolerance * s_min:
        raise BoundaryMergeRefused(
            f"dispersion {dispersion:.1f}s exceeds "
            f"{dispersion_tolerance} * S_min ({s_min:.1f}s); "
            f"escalate to parent index level"
        )

    # Safe: fit on the mean timestamp; bias is below tolerance.
    t_mean = sum(tr.t for tr in traces) / len(traces)
    r_mean = sum(math.exp(-(tr.t - window_start) / tr.stability) for tr in traces) / len(traces)
    if r_mean <= 0.0 or r_mean >= 1.0:
        raise BoundaryMergeRefused(f"degenerate mean retention {r_mean}")
    s_eff = (t_mean - window_start) / -math.log(r_mean)
    return s_eff