X-Financial/server/tests/test_agent_asset_release_recall.py

from __future__ import annotations

import pytest

from app.services.agent_asset_release_guard import ReleaseEvaluationInput
from app.services.agent_asset_release_policy import (
    ReleaseGuardPolicy,
    evaluate_release,
)
from app.services.agent_asset_release_recall import estimate_release_recall


def test_recall_uses_random_stratum_to_estimate_full_negative_population() -> None:
    result = estimate_release_recall(
        true_positive_count=90,
        disagreement_false_negative_count=2,
        random_negative_population_count=800,
        random_reviewed_count=80,
        random_false_negative_count=1,
    )

    assert result.estimated_false_negative_count == 12.0
    assert result.recall == pytest.approx(90 / 102, abs=1e-6)
    assert result.recall_lower_bound is not None
    assert result.recall_lower_bound < result.recall
    assert result.false_negative_upper_bound is not None
    assert result.false_negative_upper_bound > result.estimated_false_negative_count
    assert result.method == "stratified_random_audit_wilson_upper_bound"


def test_recall_is_exact_when_no_random_negative_stratum_exists() -> None:
    result = estimate_release_recall(
        true_positive_count=18,
        disagreement_false_negative_count=2,
        random_negative_population_count=0,
        random_reviewed_count=0,
        random_false_negative_count=0,
    )

    assert result.recall == 0.9
    assert result.recall_lower_bound == 0.9
    assert result.estimated_false_negative_count == 2.0


def test_recall_remains_unavailable_before_random_audit_evidence_exists() -> None:
    result = estimate_release_recall(
        true_positive_count=10,
        disagreement_false_negative_count=0,
        random_negative_population_count=25,
        random_reviewed_count=0,
        random_false_negative_count=0,
    )

    assert result.recall is None
    assert result.recall_lower_bound is None
    assert result.estimated_false_negative_count is None


def test_zero_true_positives_with_confirmed_miss_has_zero_recall() -> None:
    result = estimate_release_recall(
        true_positive_count=0,
        disagreement_false_negative_count=1,
        random_negative_population_count=0,
        random_reviewed_count=0,
        random_false_negative_count=0,
    )

    assert result.recall == 0.0
    assert result.recall_lower_bound == 0.0


@pytest.mark.parametrize(
    "kwargs",
    [
        {"random_reviewed_count": 3, "random_negative_population_count": 2},
        {"random_false_negative_count": 2, "random_reviewed_count": 1},
        {"true_positive_count": -1},
        {"confidence_level": 0.92},
    ],
)
def test_recall_rejects_inconsistent_evidence(kwargs: dict[str, int | float]) -> None:
    values: dict[str, int | float] = {
        "true_positive_count": 1,
        "disagreement_false_negative_count": 0,
        "random_negative_population_count": 2,
        "random_reviewed_count": 1,
        "random_false_negative_count": 0,
        "confidence_level": 0.95,
    }
    values.update(kwargs)

    with pytest.raises(ValueError):
        estimate_release_recall(**values)  # type: ignore[arg-type]


def test_release_gate_collects_without_blind_negative_ground_truth() -> None:
    result = evaluate_release(
        "shadow",
        ReleaseEvaluationInput(
            total=20,
            failure_count=0,
            precision=1.0,
            details={
                "metric_source": "release_runtime_telemetry",
                "negative_ground_truth_status": "insufficient_random_negative_reviews",
                "recall_lower_bound": None,
            },
        ),
        ReleaseGuardPolicy(shadow_min_samples=1).to_dict(),
    )

    assert result["status"] == "collecting"
    assert result["reasons"] == ["insufficient_random_negative_reviews"]


def test_release_gate_uses_recall_confidence_lower_bound_not_point_estimate() -> None:
    policy = ReleaseGuardPolicy(shadow_min_samples=1, min_recall=0.95).to_dict()
    result = evaluate_release(
        "shadow",
        ReleaseEvaluationInput(
            total=20,
            failure_count=0,
            precision=1.0,
            details={
                "metric_source": "release_runtime_telemetry",
                "negative_ground_truth_status": "available_stratified_random_audit",
                "recall": 0.98,
                "recall_lower_bound": 0.9,
            },
        ),
        policy,
    )

    assert result["status"] == "failed"
    assert "recall_lower_bound_below_threshold" in result["reasons"]


def test_release_gate_passes_when_precision_recall_and_samples_are_sufficient() -> None:
    result = evaluate_release(
        "shadow",
        ReleaseEvaluationInput(
            total=20,
            failure_count=0,
            precision=0.99,
            details={
                "metric_source": "release_runtime_telemetry",
                "negative_ground_truth_status": "available_stratified_random_audit",
                "recall": 0.98,
                "recall_lower_bound": 0.96,
            },
        ),
        ReleaseGuardPolicy(shadow_min_samples=20, min_recall=0.95).to_dict(),
    )

    assert result["status"] == "passed"