330 lines
13 KiB
Python
330 lines
13 KiB
Python
from __future__ import annotations
|
|
|
|
import copy
|
|
import unittest
|
|
|
|
from pydantic import ValidationError
|
|
|
|
from .contracts.g7_external_evidence import G7HumanVoiceGainEvidencePack
|
|
from .services.g7_voice_gain_evidence import (
|
|
VoiceGainEvidenceThresholds,
|
|
evaluate_human_voice_gain,
|
|
)
|
|
|
|
|
|
def _sha(character: str) -> str:
|
|
return character * 64
|
|
|
|
|
|
def _valid_payload() -> dict[str, object]:
|
|
participants = [
|
|
{
|
|
"participant_key": "calibration-001",
|
|
"split": "calibration",
|
|
"consent_receipt_sha256": _sha("1"),
|
|
},
|
|
{
|
|
"participant_key": "held-out-001",
|
|
"split": "held_out",
|
|
"consent_receipt_sha256": _sha("2"),
|
|
},
|
|
{
|
|
"participant_key": "held-out-002",
|
|
"split": "held_out",
|
|
"consent_receipt_sha256": _sha("3"),
|
|
},
|
|
]
|
|
observations: list[dict[str, object]] = []
|
|
targets = {
|
|
"goal": (0.20, "low"),
|
|
"task": (0.50, "medium"),
|
|
"bond": (0.80, "high"),
|
|
}
|
|
observation_number = 0
|
|
for participant_number in (1, 2):
|
|
participant = f"held-out-{participant_number:03d}"
|
|
session = f"session-{participant_number:03d}"
|
|
for axis, (target, category) in targets.items():
|
|
observation_number += 1
|
|
observations.append(
|
|
{
|
|
"observation_id": (
|
|
f"g7-human-observation-{observation_number:03d}"
|
|
),
|
|
"participant_key": participant,
|
|
"session_key": session,
|
|
"axis": axis,
|
|
"text_only_status": "observed",
|
|
"text_only_score": target + 0.20,
|
|
"voice_enabled_status": "observed",
|
|
"voice_enabled_score": target + 0.05,
|
|
"labels": [
|
|
{
|
|
"labeler_key": "labeler-001",
|
|
"score": target,
|
|
"category": category,
|
|
},
|
|
{
|
|
"labeler_key": "labeler-002",
|
|
"score": target,
|
|
"category": category,
|
|
},
|
|
],
|
|
}
|
|
)
|
|
return {
|
|
"provenance": {
|
|
"protocol_sha256": _sha("a"),
|
|
"consent_protocol_sha256": _sha("b"),
|
|
"dataset_manifest_sha256": _sha("c"),
|
|
"split_manifest_sha256": _sha("d"),
|
|
"labeling_protocol_sha256": _sha("e"),
|
|
"analysis_plan_sha256": _sha("f"),
|
|
"registered_at": "2026-08-01T00:00:00Z",
|
|
"held_out_labels_opened_at": "2026-08-02T00:00:00Z",
|
|
},
|
|
"text_only_model": {
|
|
"role": "text_only_baseline",
|
|
"provider": "example-provider",
|
|
"model_id": "alliance-text",
|
|
"model_version": "v1",
|
|
"artifact_sha256": _sha("4"),
|
|
"configuration_sha256": _sha("5"),
|
|
},
|
|
"voice_enabled_model": {
|
|
"role": "voice_enabled_candidate",
|
|
"provider": "example-provider",
|
|
"model_id": "alliance-voice",
|
|
"model_version": "v2",
|
|
"artifact_sha256": _sha("6"),
|
|
"configuration_sha256": _sha("7"),
|
|
},
|
|
"power_plan": {
|
|
"required_held_out_participants": 2,
|
|
"required_held_out_sessions": 2,
|
|
"required_paired_axis_observations": 6,
|
|
"alpha": 0.05,
|
|
"target_power": 0.8,
|
|
"minimally_detectable_gain": 0.01,
|
|
},
|
|
"participants": participants,
|
|
"labeler_attestations": [
|
|
{
|
|
"labeler_key": "labeler-001",
|
|
"attestation_sha256": _sha("8"),
|
|
},
|
|
{
|
|
"labeler_key": "labeler-002",
|
|
"attestation_sha256": _sha("9"),
|
|
},
|
|
],
|
|
"reliability": {
|
|
"labeler_keys": ["labeler-001", "labeler-002"],
|
|
"reported_icc": 1.0,
|
|
"reported_categorical_kappa": 1.0,
|
|
"report_sha256": _sha("0"),
|
|
},
|
|
"observations": observations,
|
|
}
|
|
|
|
|
|
def _test_thresholds() -> VoiceGainEvidenceThresholds:
|
|
return VoiceGainEvidenceThresholds.for_test(
|
|
min_held_out_participants=2,
|
|
min_held_out_sessions=2,
|
|
min_paired_axis_observations=6,
|
|
bootstrap_samples=500,
|
|
)
|
|
|
|
|
|
class G7HumanVoiceGainEvidenceTests(unittest.TestCase):
|
|
def test_valid_human_holdout_recomputes_paired_gain_and_cluster_ci(self) -> None:
|
|
pack = G7HumanVoiceGainEvidencePack.model_validate(_valid_payload())
|
|
|
|
result = evaluate_human_voice_gain(pack, thresholds=_test_thresholds())
|
|
|
|
self.assertTrue(result.passed, result.failure_reasons)
|
|
self.assertAlmostEqual(result.text_only_one_minus_mae, 0.8)
|
|
self.assertAlmostEqual(result.voice_enabled_one_minus_mae, 0.95)
|
|
self.assertAlmostEqual(result.paired_gain, 0.15)
|
|
self.assertGreater(result.ci_lower, 0.0)
|
|
self.assertEqual(result.recomputed_icc, 1.0)
|
|
self.assertEqual(result.recomputed_categorical_kappa, 1.0)
|
|
self.assertFalse(result.clinical_claim_allowed)
|
|
|
|
def test_production_defaults_reject_small_fixture(self) -> None:
|
|
pack = G7HumanVoiceGainEvidencePack.model_validate(_valid_payload())
|
|
|
|
result = evaluate_human_voice_gain(pack)
|
|
|
|
self.assertFalse(result.passed)
|
|
self.assertIn("production_participant_floor", result.failure_reasons)
|
|
self.assertIn("production_session_floor", result.failure_reasons)
|
|
self.assertIn("production_observation_floor", result.failure_reasons)
|
|
self.assertIn("power_plan_participant_floor", result.failure_reasons)
|
|
self.assertIn("power_plan_session_floor", result.failure_reasons)
|
|
self.assertIn("power_plan_observation_floor", result.failure_reasons)
|
|
self.assertEqual(result.bootstrap_samples, 10_000)
|
|
|
|
def test_missing_or_error_prediction_is_max_error_intention_to_evaluate(self) -> None:
|
|
payload = _valid_payload()
|
|
observations = payload["observations"]
|
|
assert isinstance(observations, list)
|
|
observations[0]["voice_enabled_status"] = "error"
|
|
observations[0]["voice_enabled_score"] = None
|
|
pack = G7HumanVoiceGainEvidencePack.model_validate(payload)
|
|
|
|
result = evaluate_human_voice_gain(pack, thresholds=_test_thresholds())
|
|
|
|
self.assertEqual(result.intention_to_evaluate_imputations, 1)
|
|
expected_voice_accuracy = 1.0 - ((1.0 + (0.05 * 5)) / 6)
|
|
self.assertAlmostEqual(result.voice_enabled_one_minus_mae, expected_voice_accuracy)
|
|
self.assertEqual(result.paired_axis_observations, 5)
|
|
self.assertEqual(result.held_out_sessions, 1)
|
|
self.assertIn("production_observation_floor", result.failure_reasons)
|
|
self.assertIn("production_session_floor", result.failure_reasons)
|
|
self.assertFalse(result.passed)
|
|
|
|
def test_baseline_only_missing_cannot_inflate_candidate_gain(self) -> None:
|
|
complete = G7HumanVoiceGainEvidencePack.model_validate(_valid_payload())
|
|
complete_result = evaluate_human_voice_gain(
|
|
complete,
|
|
thresholds=_test_thresholds(),
|
|
)
|
|
payload = _valid_payload()
|
|
observations = payload["observations"]
|
|
assert isinstance(observations, list)
|
|
observations[0]["text_only_status"] = "missing"
|
|
observations[0]["text_only_score"] = None
|
|
missing = G7HumanVoiceGainEvidencePack.model_validate(payload)
|
|
|
|
missing_result = evaluate_human_voice_gain(
|
|
missing,
|
|
thresholds=_test_thresholds(),
|
|
)
|
|
|
|
self.assertLess(missing_result.paired_gain, complete_result.paired_gain)
|
|
self.assertEqual(missing_result.intention_to_evaluate_imputations, 1)
|
|
self.assertEqual(missing_result.paired_axis_observations, 5)
|
|
|
|
def test_duplicate_or_calibration_observation_is_rejected(self) -> None:
|
|
duplicate = _valid_payload()
|
|
duplicate_rows = duplicate["observations"]
|
|
assert isinstance(duplicate_rows, list)
|
|
duplicate_rows.append(copy.deepcopy(duplicate_rows[0]))
|
|
with self.assertRaisesRegex(ValidationError, "observation ids must be unique"):
|
|
G7HumanVoiceGainEvidencePack.model_validate(duplicate)
|
|
|
|
leakage = _valid_payload()
|
|
leakage_rows = leakage["observations"]
|
|
assert isinstance(leakage_rows, list)
|
|
leakage_rows[0]["participant_key"] = "calibration-001"
|
|
with self.assertRaisesRegex(ValidationError, "held-out participants"):
|
|
G7HumanVoiceGainEvidencePack.model_validate(leakage)
|
|
|
|
def test_blind_independent_attestation_and_full_axis_coverage_are_required(self) -> None:
|
|
unblinded = _valid_payload()
|
|
attestations = unblinded["labeler_attestations"]
|
|
assert isinstance(attestations, list)
|
|
attestations[0]["blinded_to_model_condition"] = False
|
|
with self.assertRaises(ValidationError):
|
|
G7HumanVoiceGainEvidencePack.model_validate(unblinded)
|
|
|
|
incomplete = _valid_payload()
|
|
rows = incomplete["observations"]
|
|
assert isinstance(rows, list)
|
|
rows.pop()
|
|
with self.assertRaisesRegex(ValidationError, "goal, task, and bond"):
|
|
G7HumanVoiceGainEvidencePack.model_validate(incomplete)
|
|
|
|
def test_raw_material_and_synthetic_pack_cannot_enter_contract(self) -> None:
|
|
raw = _valid_payload()
|
|
rows = raw["observations"]
|
|
assert isinstance(rows, list)
|
|
rows[0]["raw_audio"] = "forbidden"
|
|
with self.assertRaises(ValidationError):
|
|
G7HumanVoiceGainEvidencePack.model_validate(raw)
|
|
|
|
synthetic = _valid_payload()
|
|
synthetic["synthetic_pack"] = True
|
|
with self.assertRaises(ValidationError):
|
|
G7HumanVoiceGainEvidencePack.model_validate(synthetic)
|
|
|
|
def test_reported_reliability_is_checked_against_rows(self) -> None:
|
|
payload = _valid_payload()
|
|
reliability = payload["reliability"]
|
|
assert isinstance(reliability, dict)
|
|
reliability["reported_icc"] = 0.8
|
|
pack = G7HumanVoiceGainEvidencePack.model_validate(payload)
|
|
|
|
result = evaluate_human_voice_gain(pack, thresholds=_test_thresholds())
|
|
|
|
self.assertFalse(result.passed)
|
|
self.assertIn("reported_icc_matches_rows", result.failure_reasons)
|
|
|
|
def test_categorical_kappa_is_mandatory_and_recomputed(self) -> None:
|
|
missing_report = _valid_payload()
|
|
reliability = missing_report["reliability"]
|
|
assert isinstance(reliability, dict)
|
|
reliability.pop("reported_categorical_kappa")
|
|
with self.assertRaises(ValidationError):
|
|
G7HumanVoiceGainEvidencePack.model_validate(missing_report)
|
|
|
|
missing_category = _valid_payload()
|
|
observations = missing_category["observations"]
|
|
assert isinstance(observations, list)
|
|
labels = observations[0]["labels"]
|
|
assert isinstance(labels, list)
|
|
labels[0].pop("category")
|
|
with self.assertRaises(ValidationError):
|
|
G7HumanVoiceGainEvidencePack.model_validate(missing_category)
|
|
|
|
mismatched = _valid_payload()
|
|
reliability = mismatched["reliability"]
|
|
assert isinstance(reliability, dict)
|
|
reliability["reported_categorical_kappa"] = 0.8
|
|
pack = G7HumanVoiceGainEvidencePack.model_validate(mismatched)
|
|
result = evaluate_human_voice_gain(pack, thresholds=_test_thresholds())
|
|
self.assertIn("reported_kappa_matches_rows", result.failure_reasons)
|
|
|
|
weak = _valid_payload()
|
|
observations = weak["observations"]
|
|
assert isinstance(observations, list)
|
|
for observation in observations:
|
|
labels = observation["labels"]
|
|
assert isinstance(labels, list)
|
|
labels[0]["category"] = "low"
|
|
labels[1]["category"] = "high"
|
|
reliability = weak["reliability"]
|
|
assert isinstance(reliability, dict)
|
|
reliability["reported_categorical_kappa"] = -1.0
|
|
weak_pack = G7HumanVoiceGainEvidencePack.model_validate(weak)
|
|
weak_result = evaluate_human_voice_gain(
|
|
weak_pack,
|
|
thresholds=_test_thresholds(),
|
|
)
|
|
self.assertIn("recomputed_categorical_kappa", weak_result.failure_reasons)
|
|
|
|
def test_analysis_registration_must_strictly_precede_held_out_access(self) -> None:
|
|
payload = _valid_payload()
|
|
provenance = payload["provenance"]
|
|
assert isinstance(provenance, dict)
|
|
provenance["registered_at"] = provenance["held_out_labels_opened_at"]
|
|
with self.assertRaisesRegex(ValidationError, "must precede"):
|
|
G7HumanVoiceGainEvidencePack.model_validate(payload)
|
|
|
|
wrong_alpha = _valid_payload()
|
|
power_plan = wrong_alpha["power_plan"]
|
|
assert isinstance(power_plan, dict)
|
|
power_plan["alpha"] = 0.01
|
|
with self.assertRaises(ValidationError):
|
|
G7HumanVoiceGainEvidencePack.model_validate(wrong_alpha)
|
|
|
|
def test_custom_thresholds_require_explicit_test_factory(self) -> None:
|
|
pack = G7HumanVoiceGainEvidencePack.model_validate(_valid_payload())
|
|
with self.assertRaisesRegex(ValueError, "test-only"):
|
|
evaluate_human_voice_gain(pack, thresholds=VoiceGainEvidenceThresholds())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|