NODEDC_MISSION_CORE/tests/test_e41_evaluation_boundary.py

from __future__ import annotations

import numpy as np
import pytest

from k1link.compute.e41_evaluation_boundary import (
    E41EvaluationBoundaryError,
    evaluate_visible_engineering_contract,
    predict_from_frozen_e40_model,
)


def _model() -> dict[str, object]:
    return {
        "schema_version": "missioncore.e40-development-product-model/v1",
        "feature_names": ["signal"],
        "validation_labels_used_for_training": False,
        "robust_clip": 10.0,
        "fixed_presence_by_stratum": {
            "agree": "object-present",
            "conflict": "background-or-noise",
            "geometry-only": "occupied-environment",
            "unknown": "object-present",
        },
        "dimension_projection": "source-stratum-plus-presence/v1",
        "classifier": {
            "type": "deterministic-softmax",
            "labels": [
                "background-or-noise",
                "object-present",
                "occupied-environment",
            ],
            "weights": [
                [-2.0, 2.0, 0.0],
                [0.0, 0.0, 0.0],
            ],
            "scaler": {
                "median": [0.0],
                "scale": [1.0],
            },
        },
    }


def test_e41_predictor_output_is_truth_free_and_deterministic() -> None:
    items = [
        {
            "schema_version": "missioncore.e41-predictor-item/v1",
            "sequence": 0,
            "item_id": "a",
            "source_stratum": "camera-only",
        },
        {
            "schema_version": "missioncore.e41-predictor-item/v1",
            "sequence": 1,
            "item_id": "b",
            "source_stratum": "geometry-only",
        },
    ]
    matrix = np.asarray([[1.0], [0.0]], dtype=np.float64)
    first = predict_from_frozen_e40_model(
        items=items,
        feature_names=["signal"],
        feature_matrix=matrix,
        model=_model(),
    )
    second = predict_from_frozen_e40_model(
        items=items,
        feature_names=["signal"],
        feature_matrix=matrix,
        model=_model(),
    )

    assert first == second
    assert first[0]["prediction"]["presence"] == "object-present"
    assert first[1]["prediction"]["presence"] == "occupied-environment"
    assert not any(
        forbidden in row
        for row in first
        for forbidden in ("reference", "scored", "severity", "split", "truth")
    )


def test_e41_predictor_rejects_truth_bearing_item_metadata() -> None:
    with pytest.raises(E41EvaluationBoundaryError, match="truth/evaluation"):
        predict_from_frozen_e40_model(
            items=[
                {
                    "sequence": 0,
                    "item_id": "a",
                    "source_stratum": "camera-only",
                    "reference": {"presence": "object-present"},
                }
            ],
            feature_names=["signal"],
            feature_matrix=np.asarray([[1.0]], dtype=np.float64),
            model=_model(),
        )


def test_e41_visible_evaluator_joins_truth_after_prediction() -> None:
    predictions = [
        {
            "item_id": "dev",
            "prediction": {
                "presence": "object-present",
                "geometry_association": "insufficient-support",
                "freshness": "unavailable",
            },
        },
        {
            "item_id": "val",
            "prediction": {
                "presence": "object-present",
                "geometry_association": "object-associated",
                "freshness": "current",
            },
        },
    ]
    acceptance = [
        {
            "item_id": "dev",
            "split": "development",
            "source_stratum": "camera-only",
            "severity": "standard",
            "reference": {
                "presence": "background-or-noise",
                "geometry_association": "rejected-nonobject",
                "freshness": "unavailable",
            },
        },
        {
            "item_id": "val",
            "split": "validation",
            "source_stratum": "agree",
            "severity": "standard",
            "reference": {
                "presence": "object-present",
                "geometry_association": "object-associated",
                "freshness": "current",
            },
        },
    ]
    evaluation = evaluate_visible_engineering_contract(
        predictions=predictions,
        acceptance_rows=acceptance,
        targets={
            "presence_target": 0.9,
            "geometry_association_target": 0.9,
            "freshness_target": 0.9,
        },
        label_provenance={
            "engineering_items": 2,
            "human_exception_items": 0,
            "independent_ground_truth": False,
        },
    )

    assert evaluation["metrics"]["validation_items"] == 1
    assert evaluation["metrics"]["dimensions"]["presence"]["accuracy"] == 1.0
    assert evaluation["engineering_contract_targets_reached"] is True
    assert evaluation["blind_gate_eligible"] is False
    assert evaluation["label_provenance"]["independent_accuracy_authority"] is False