tests/test_retro/test_agent.py

"""Agent pipeline tests: canonical fixtures, fact/inference separation,
stripped-telemetry -> UNKNOWN, scorecard completeness and tier tagging."""

from __future__ import annotations

import json

import pytest

from sensorflow.retro import store as retro_store
from sensorflow.retro.agent.orchestrator import analyze, analyze_fixture
from sensorflow.retro.scorecard import (EvidenceTier, LaunchRecommendation,
                                        RetrospectiveScorecard, Severity)


@pytest.fixture()
def phantom(retro_root) -> RetrospectiveScorecard:
    return analyze_fixture("phantom_brake_plastic_bag", backend="mock")


@pytest.fixture()
def missed_ped(retro_root) -> RetrospectiveScorecard:
    return analyze_fixture("missed_pedestrian_rain", backend="mock")


# ----------------------------------------------------- canonical fixture A

def test_phantom_brake_fixture_analysis(phantom):
    sc = phantom
    assert sc.failure_type == "FALSE_POSITIVE"
    assert sc.severity == Severity.CRITICAL          # hard brake + tailgater
    assert sc.launch_recommendation == LaunchRecommendation.FAIL
    assert sc.object_class == "plastic_bag"
    # causal hypotheses reference the actual evidence
    texts = " ".join(h.hypothesis.lower() for h in sc.root_cause_hypotheses)
    assert "deformable" in texts or "training-distribution" in texts
    assert any("velocity latch" in h.hypothesis.lower()
               for h in sc.root_cause_hypotheses)
    # phantom braking requirement was retrieved
    assert any(s.doc_id == "SFS-SAFE-001" for s in sc.retrieved_standards)


# ----------------------------------------------------- canonical fixture B

def test_missed_pedestrian_fixture_analysis(missed_ped):
    sc = missed_ped
    assert sc.failure_type == "FALSE_NEGATIVE"
    assert sc.severity == Severity.CRITICAL
    assert sc.launch_recommendation == LaunchRecommendation.FAIL
    assert sc.human_review_required
    # kinematics: pedestrian inside stopping distance
    assert sc.stopping_distance_m is not None
    assert sc.distance_to_object_m <= sc.stopping_distance_m * 1.05
    assert sc.ttc_s is not None and sc.ttc_s < 2.5
    # rain-conditioned hypothesis grounded in sensor evidence
    assert any("fusion" in h.hypothesis.lower() for h in sc.root_cause_hypotheses)
    # SCR regression computed and significant -> reflected in the gate
    assert sc.safety_critical_recall_impact == pytest.approx(-3 / 1240, abs=1e-6)
    assert sc.statistical_significance.significant is True
    assert any(s.doc_id in ("SFS-SAFE-001", "SFS-PERC-002")
               for s in sc.retrieved_standards)


# --------------------------------------------------- fact/inference separation

def test_fact_inference_separation(missed_ped):
    sc = missed_ped
    tiers = {e.tier for e in sc.evidence}
    assert {EvidenceTier.OBSERVED, EvidenceTier.DERIVED,
            EvidenceTier.RETRIEVED, EvidenceTier.AI_HYPOTHESIS} <= tiers
    # hypotheses never appear as observed/derived facts
    for e in sc.evidence:
        if e.tier == EvidenceTier.AI_HYPOTHESIS:
            assert "inference" in e.provenance
    for h in sc.root_cause_hypotheses:
        assert h.tier == EvidenceTier.AI_HYPOTHESIS
        assert h.supporting_evidence_keys, "hypotheses must cite evidence"
    # observed items carry artifact provenance
    observed = [e for e in sc.evidence if e.tier == EvidenceTier.OBSERVED]
    assert all("artifact" in e.provenance or "eval log field" in e.provenance
               for e in observed)
    # rendered scorecard keeps the five-way distinction
    md = sc.render_markdown()
    for heading in ("FACT", "DERIVED FACT", "RETRIEVED REQUIREMENT",
                    "AI HYPOTHES", "SAFETY DETERMINATION"):
        assert heading in md


# ------------------------------------------------- stripped telemetry -> UNKNOWN

def test_stripped_telemetry_produces_unknown_not_guesses(retro_root):
    sc = analyze_fixture("missing_telemetry_variant", backend="mock")
    assert sc.failure_type == "FALSE_NEGATIVE"
    # stripped fields must be UNKNOWN, never guessed
    assert sc.ego_speed_mps is None
    assert sc.stopping_distance_m is None
    assert sc.ttc_s is None
    assert "ego.speed_mps" in sc.uncertainty.missing_fields
    assert "counterfactual_planner_response" in sc.uncertainty.missing_fields
    assert "stopping_distance" in sc.uncertainty.unknown_metrics
    assert "ttc" in sc.uncertainty.unknown_metrics
    # UNKNOWN markers surface in the evidence chain
    unknown_ev = [e for e in sc.evidence if e.value == "UNKNOWN"]
    assert unknown_ev
    # and the gate refuses to make a decision
    assert sc.launch_recommendation == LaunchRecommendation.INSUFFICIENT_EVIDENCE
    assert sc.human_review_required


def test_dynamically_stripped_field_becomes_unknown(retro_root):
    """Strip a field from fixture B at runtime -> the output marks it UNKNOWN."""
    from sensorflow.retro.tools.builtin import FIXTURES_DIR
    log = json.loads((FIXTURES_DIR / "missed_pedestrian_rain.json").read_text())
    del log["ego"]["speed_mps"]
    log["evaluation_id"] = "EVAL-STRIPPED-01"
    upload = retro_store.uploads_dir() / "stripped.json"
    upload.write_text(json.dumps(log))

    sc = analyze(path=str(upload), backend="mock")
    assert sc.ego_speed_mps is None
    assert "ego.speed_mps" in sc.uncertainty.missing_fields
    assert sc.stopping_distance_m is None  # not computable without speed
    assert sc.launch_recommendation == LaunchRecommendation.INSUFFICIENT_EVIDENCE


# ------------------------------------------------- severity divergence flow

def test_llm_severity_divergence_flagged(retro_root):
    """On the missing-telemetry variant the scripted 'LLM' proposes CRITICAL
    (near-miss heuristic) while policy computes DISRUPTIVE from the sparse
    evidence -> divergence recorded, policy wins, human review required."""
    sc = analyze_fixture("missing_telemetry_variant", backend="mock")
    assert sc.ai_proposed_severity == Severity.CRITICAL
    assert sc.severity == Severity.DISRUPTIVE
    assert sc.severity_divergence
    assert any("diverged" in r for r in sc.human_review_reasons)
    assert any("severity adjudication" in n for n in sc.uncertainty.notes)


# ------------------------------------------------------ variants + persistence

def test_benign_variants_pass(retro_root):
    fp = analyze_fixture("benign_fp_static_sign", backend="mock")
    fn = analyze_fixture("noncritical_fn_parked_vehicle", backend="mock")
    assert fp.severity == Severity.BENIGN
    assert fp.launch_recommendation == LaunchRecommendation.PASS
    assert fn.severity == Severity.BENIGN
    assert fn.launch_recommendation == LaunchRecommendation.PASS
    assert not fn.human_review_required


def test_analysis_persisted_with_audit(retro_root):
    sc = analyze_fixture("phantom_brake_plastic_bag", backend="mock")
    stored = retro_store.load_analysis(sc.evaluation_id)
    assert stored and stored["scorecard"]["failure_type"] == "FALSE_POSITIVE"
    assert "# Retrospective Scorecard" in stored["markdown"]
    audit = retro_store.read_audit(stored["audit_analysis_id"])
    tools_used = {r["tool"] for r in audit}
    assert {"log_reader", "metric_calculator", "safety_standard_rag",
            "historical_failure_search", "distribution_analysis"} <= tools_used
    # no write-tool call happened during analysis
    assert "create_evaluation_case" not in tools_used


def test_scorecard_completeness_spec_fields(missed_ped):
    """Every spec field exists and the strictly-typed model round-trips."""
    sc = missed_ped
    dumped = json.loads(sc.model_dump_json())
    for field in ["evaluation_id", "failure_type", "severity",
                  "safety_critical_recall_impact", "behavioral_consequence",
                  "launch_recommendation", "baseline_model", "candidate_model",
                  "scenario", "object_class", "ground_truth", "prediction",
                  "confidence", "ego_speed_mps", "distance_to_object_m",
                  "stopping_distance_m", "ttc_s", "planner_response",
                  "disengagement_probability", "metric_delta",
                  "statistical_significance", "distribution_shift",
                  "root_cause_hypotheses", "evidence", "retrieved_standards",
                  "uncertainty", "human_review_required", "policy_version",
                  "agent_version"]:
        assert field in dumped, f"missing spec field {field}"
    assert dumped["policy_version"].startswith("retro-policy/")
    assert dumped["agent_version"].startswith("retro-agent/")
    restored = RetrospectiveScorecard(**dumped)
    assert restored.severity == sc.severity


def test_historical_search_finds_prior_analyses(retro_root):
    analyze_fixture("missed_pedestrian_rain", backend="mock")
    from sensorflow.retro.tools.builtin import build_registry
    reg = build_registry(analysis_id="hist-test")
    res = reg.call("historical_failure_search",
                   {"query_text": "pedestrian rain night missed detection", "k": 3})
    assert res.result["corpus_size"] >= 1
    assert any(m["evaluation_id"] == "EVAL-2026-0802-MP01"
               for m in res.result["matches"])