"""Fail-closed evaluator for independent human-labeled G7 voice gain.""" from __future__ import annotations import math import random from collections import Counter, defaultdict from dataclasses import dataclass, replace from typing import Literal from pydantic import BaseModel, ConfigDict from ..contracts.g7_external_evidence import G7HumanVoiceGainEvidencePack @dataclass(frozen=True, slots=True) class VoiceGainEvidenceThresholds: min_held_out_participants: int = 30 min_held_out_sessions: int = 50 min_paired_axis_observations: int = 150 min_icc: float = 0.75 min_categorical_kappa: float = 0.70 min_gain: float = 0.01 bootstrap_samples: int = 10_000 confidence_level: float = 0.95 seed: int = 20260807 test_only: bool = False @classmethod def for_test(cls, **overrides: object) -> "VoiceGainEvidenceThresholds": """Create an explicit small-fixture override that production cannot imply.""" return replace(cls(), **overrides, test_only=True) PRODUCTION_THRESHOLDS = VoiceGainEvidenceThresholds() class VoiceGainEvidenceCheck(BaseModel): model_config = ConfigDict(extra="forbid", frozen=True) name: str passed: bool actual: int | float | str requirement: int | float | str class VoiceGainEvidenceResult(BaseModel): model_config = ConfigDict(extra="forbid", frozen=True) passed: bool clinical_claim_allowed: Literal[False] = False held_out_participants: int held_out_sessions: int paired_axis_observations: int intention_to_evaluate_imputations: int text_only_one_minus_mae: float voice_enabled_one_minus_mae: float paired_gain: float ci_lower: float ci_upper: float confidence_level: float bootstrap_samples: int recomputed_icc: float recomputed_categorical_kappa: float | None checks: tuple[VoiceGainEvidenceCheck, ...] failure_reasons: tuple[str, ...] def _icc_absolute_agreement_single(ratings: list[list[float]]) -> float: """Calculate balanced two-way random absolute-agreement ICC(A,1).""" target_count = len(ratings) rater_count = len(ratings[0]) if ratings else 0 if target_count < 2 or rater_count < 2: raise ValueError("ICC requires at least two targets and two labelers") if any(len(row) != rater_count for row in ratings): raise ValueError("ICC requires a balanced label matrix") grand_mean = sum(sum(row) for row in ratings) / (target_count * rater_count) row_means = [sum(row) / rater_count for row in ratings] column_means = [ sum(row[index] for row in ratings) / target_count for index in range(rater_count) ] ms_rows = rater_count * sum( (mean - grand_mean) ** 2 for mean in row_means ) / (target_count - 1) ms_columns = target_count * sum( (mean - grand_mean) ** 2 for mean in column_means ) / (rater_count - 1) residual = sum( ( ratings[row_index][column_index] - row_means[row_index] - column_means[column_index] + grand_mean ) ** 2 for row_index in range(target_count) for column_index in range(rater_count) ) ms_error = residual / ((target_count - 1) * (rater_count - 1)) denominator = ( ms_rows + (rater_count - 1) * ms_error + rater_count * (ms_columns - ms_error) / target_count ) if math.isclose(denominator, 0.0, abs_tol=1e-15): raise ValueError("ICC is undefined for a zero-variance label matrix") return max(-1.0, min(1.0, (ms_rows - ms_error) / denominator)) def _fleiss_kappa(categories: list[list[str]]) -> float: target_count = len(categories) rater_count = len(categories[0]) if categories else 0 if target_count < 2 or rater_count < 2: raise ValueError("categorical kappa requires two targets and labelers") if any(len(row) != rater_count for row in categories): raise ValueError("categorical kappa requires a balanced label matrix") category_names = sorted({value for row in categories for value in row}) total_counts: Counter[str] = Counter() per_target_agreement: list[float] = [] for row in categories: counts = Counter(row) total_counts.update(counts) numerator = sum(count * count for count in counts.values()) - rater_count per_target_agreement.append(numerator / (rater_count * (rater_count - 1))) observed = sum(per_target_agreement) / target_count expected = sum( (total_counts[name] / (target_count * rater_count)) ** 2 for name in category_names ) if math.isclose(1.0 - expected, 0.0, abs_tol=1e-15): if math.isclose(observed, 1.0, abs_tol=1e-15): return 1.0 raise ValueError("categorical kappa is undefined") return max(-1.0, min(1.0, (observed - expected) / (1.0 - expected))) def _percentile(values: list[float], probability: float) -> float: ordered = sorted(values) position = (len(ordered) - 1) * probability lower_index = math.floor(position) upper_index = math.ceil(position) if lower_index == upper_index: return ordered[lower_index] fraction = position - lower_index return ordered[lower_index] * (1.0 - fraction) + ordered[upper_index] * fraction def _cluster_bootstrap( gains_by_participant: dict[str, list[float]], *, samples: int, confidence_level: float, seed: int, ) -> tuple[float, float]: if samples < 1: raise ValueError("bootstrap samples must be positive") participant_keys = sorted(gains_by_participant) if len(participant_keys) < 2: raise ValueError("participant-cluster bootstrap requires two participants") generator = random.Random(seed) draws: list[float] = [] for _ in range(samples): sampled_keys = [ generator.choice(participant_keys) for _ in participant_keys ] sampled_gains = [ gain for participant_key in sampled_keys for gain in gains_by_participant[participant_key] ] draws.append(sum(sampled_gains) / len(sampled_gains)) alpha = 1.0 - confidence_level return _percentile(draws, alpha / 2.0), _percentile(draws, 1.0 - alpha / 2.0) def _check( checks: list[VoiceGainEvidenceCheck], name: str, passed: bool, actual: int | float | str, requirement: int | float | str, ) -> None: checks.append( VoiceGainEvidenceCheck( name=name, passed=passed, actual=actual, requirement=requirement, ) ) def evaluate_human_voice_gain( pack: G7HumanVoiceGainEvidencePack, *, thresholds: VoiceGainEvidenceThresholds | None = None, ) -> VoiceGainEvidenceResult: """Recompute every gate from deidentified held-out rows. Custom thresholds are accepted only when explicitly created with ``VoiceGainEvidenceThresholds.for_test``. Normal callers always receive the production N, reliability, effect, and 10,000-bootstrap requirements. """ if thresholds is None: thresholds = PRODUCTION_THRESHOLDS elif not thresholds.test_only: raise ValueError("custom G7 evidence thresholds are test-only") panel = tuple(pack.reliability.labeler_keys) ratings: list[list[float]] = [] categories: list[list[str]] = [] errors_by_participant: dict[str, list[tuple[float, float]]] = defaultdict(list) imputation_count = 0 for observation in pack.observations: labels_by_key = {item.labeler_key: item for item in observation.labels} ordered_labels = [labels_by_key[labeler_key] for labeler_key in panel] ratings.append([item.score for item in ordered_labels]) if ordered_labels[0].category is not None: categories.append([str(item.category) for item in ordered_labels]) reference = sum(item.score for item in ordered_labels) / len(ordered_labels) if observation.text_only_status == "observed": assert observation.text_only_score is not None text_error = abs(observation.text_only_score - reference) else: text_error = 1.0 imputation_count += 1 if observation.voice_enabled_status == "observed": assert observation.voice_enabled_score is not None voice_error = abs(observation.voice_enabled_score - reference) else: voice_error = 1.0 imputation_count += 1 errors_by_participant[observation.participant_key].append( (text_error, voice_error) ) all_errors = [pair for pairs in errors_by_participant.values() for pair in pairs] text_mae = sum(pair[0] for pair in all_errors) / len(all_errors) voice_mae = sum(pair[1] for pair in all_errors) / len(all_errors) text_accuracy = 1.0 - text_mae voice_accuracy = 1.0 - voice_mae gain = voice_accuracy - text_accuracy gains_by_participant = { participant: [text_error - voice_error for text_error, voice_error in pairs] for participant, pairs in errors_by_participant.items() } ci_lower, ci_upper = _cluster_bootstrap( gains_by_participant, samples=thresholds.bootstrap_samples, confidence_level=thresholds.confidence_level, seed=thresholds.seed, ) recomputed_icc = _icc_absolute_agreement_single(ratings) recomputed_kappa = _fleiss_kappa(categories) if categories else None held_out_participants = len(errors_by_participant) held_out_sessions = len({item.session_key for item in pack.observations}) observation_count = len(pack.observations) checks: list[VoiceGainEvidenceCheck] = [] _check( checks, "production_participant_floor", held_out_participants >= thresholds.min_held_out_participants, held_out_participants, thresholds.min_held_out_participants, ) _check( checks, "production_session_floor", held_out_sessions >= thresholds.min_held_out_sessions, held_out_sessions, thresholds.min_held_out_sessions, ) _check( checks, "production_observation_floor", observation_count >= thresholds.min_paired_axis_observations, observation_count, thresholds.min_paired_axis_observations, ) _check( checks, "power_plan_participants_achieved", held_out_participants >= pack.power_plan.required_held_out_participants, held_out_participants, pack.power_plan.required_held_out_participants, ) _check( checks, "power_plan_participant_floor", pack.power_plan.required_held_out_participants >= thresholds.min_held_out_participants, pack.power_plan.required_held_out_participants, thresholds.min_held_out_participants, ) _check( checks, "power_plan_sessions_achieved", held_out_sessions >= pack.power_plan.required_held_out_sessions, held_out_sessions, pack.power_plan.required_held_out_sessions, ) _check( checks, "power_plan_session_floor", pack.power_plan.required_held_out_sessions >= thresholds.min_held_out_sessions, pack.power_plan.required_held_out_sessions, thresholds.min_held_out_sessions, ) _check( checks, "power_plan_observations_achieved", observation_count >= pack.power_plan.required_paired_axis_observations, observation_count, pack.power_plan.required_paired_axis_observations, ) _check( checks, "power_plan_observation_floor", pack.power_plan.required_paired_axis_observations >= thresholds.min_paired_axis_observations, pack.power_plan.required_paired_axis_observations, thresholds.min_paired_axis_observations, ) _check( checks, "power_plan_effect_matches_gate", pack.power_plan.minimally_detectable_gain <= thresholds.min_gain, pack.power_plan.minimally_detectable_gain, f"<= {thresholds.min_gain}", ) _check( checks, "recomputed_icc", recomputed_icc >= thresholds.min_icc, recomputed_icc, thresholds.min_icc, ) _check( checks, "reported_icc_matches_rows", math.isclose( recomputed_icc, pack.reliability.reported_icc, abs_tol=0.0005, ), recomputed_icc, pack.reliability.reported_icc, ) if recomputed_kappa is not None: assert pack.reliability.reported_categorical_kappa is not None _check( checks, "recomputed_categorical_kappa", recomputed_kappa >= thresholds.min_categorical_kappa, recomputed_kappa, thresholds.min_categorical_kappa, ) _check( checks, "reported_kappa_matches_rows", math.isclose( recomputed_kappa, pack.reliability.reported_categorical_kappa, abs_tol=0.0005, ), recomputed_kappa, pack.reliability.reported_categorical_kappa, ) _check(checks, "minimum_paired_gain", gain >= thresholds.min_gain, gain, thresholds.min_gain) _check(checks, "bootstrap_ci_excludes_zero", ci_lower > 0.0, ci_lower, "> 0") _check( checks, "bootstrap_sample_count", thresholds.bootstrap_samples == (10_000 if not thresholds.test_only else thresholds.bootstrap_samples), thresholds.bootstrap_samples, 10_000 if not thresholds.test_only else "test override", ) failures = tuple(check.name for check in checks if not check.passed) return VoiceGainEvidenceResult( passed=not failures, clinical_claim_allowed=False, held_out_participants=held_out_participants, held_out_sessions=held_out_sessions, paired_axis_observations=observation_count, intention_to_evaluate_imputations=imputation_count, text_only_one_minus_mae=text_accuracy, voice_enabled_one_minus_mae=voice_accuracy, paired_gain=gain, ci_lower=ci_lower, ci_upper=ci_upper, confidence_level=thresholds.confidence_level, bootstrap_samples=thresholds.bootstrap_samples, recomputed_icc=recomputed_icc, recomputed_categorical_kappa=recomputed_kappa, checks=tuple(checks), failure_reasons=failures, )