content-find-agent-new/tests/test_rule_judgment_hard_gates.py

from __future__ import annotations

from copy import deepcopy

import pytest

from content_agent.business_modules.rule_judgment.evaluator import decide
from content_agent.run_service import RunService
from content_agent.schemas import RunStartRequest
from tests.p1_helpers import FakeQueryVariantClient, REAL_SOURCE_FIXTURE


def test_hard_gate_outputs_rule_blocked_and_primary_reason(tmp_path):
    state = _state(tmp_path)
    bundle = deepcopy(state["evidence_bundles"][0])
    bundle["content"]["platform_content_id"] = ""
    bundle["source_evidence"] = {}

    decision = decide(state["run_id"], state["policy_run_id"], 1, bundle, state["policy_bundle"])

    assert decision["decision_action"] == "REJECT_CONTENT"
    assert decision["decision_reason_code"] == "missing_platform_content_id"
    assert decision["search_query_effect_status"] == "rule_blocked"
    assert set(decision["triggered_blocking_rules"]) >= {
        "missing_platform_content_id",
        "missing_source_evidence",
    }
    replay = decision["decision_replay_data"]
    assert replay["primary_gate_id"] == "missing_platform_content_id"
    assert replay["primary_reason_code"] == "missing_platform_content_id"
    assert replay["effect_mapping_id"] == "map_reject_rule_blocked"


def test_unknown_hard_gate_operator_fails_fast(tmp_path):
    state = _state(tmp_path)
    policy_bundle = deepcopy(state["policy_bundle"])
    policy_bundle["rule_pack"]["hard_gates"][0]["when"]["op"] = "starts_with"

    with pytest.raises(ValueError, match="unsupported rule operator"):
        decide(
            state["run_id"],
            state["policy_run_id"],
            1,
            state["evidence_bundles"][0],
            policy_bundle,
        )


def test_legacy_senior_fit_field_no_longer_triggers_hard_gate(tmp_path):
    state = _state(tmp_path)
    bundle = deepcopy(state["evidence_bundles"][0])
    bundle["pattern_match_result"]["fit_senior_50plus"] = False

    decision = decide(state["run_id"], state["policy_run_id"], 1, bundle, state["policy_bundle"])

    assert "not_fit_senior" not in decision["triggered_blocking_rules"]
    assert decision["decision_reason_code"] != "content_not_fit_senior"
    assert decision["decision_reason_code"] != "content_pattern_recall_required"


def test_hard_gate_action_is_taken_from_rule_config(tmp_path):
    # The judge_failed gate defaults to TECHNICAL_RETRY_REQUIRED; flipping its config to
    # REJECT_CONTENT must change the decision purely via config, with no code special-case.
    state = _state(tmp_path)
    bundle = deepcopy(state["evidence_bundles"][0])
    bundle["pattern_match_result"]["judge_status"] = "failed"

    decision = decide(state["run_id"], state["policy_run_id"], 1, bundle, state["policy_bundle"])
    assert decision["decision_action"] == "TECHNICAL_RETRY_REQUIRED"

    flipped = deepcopy(state["policy_bundle"])
    for gate in flipped["rule_pack"]["hard_gates"]:
        if gate["gate_id"] == "judge_failed":
            gate["decision_action"] = "REJECT_CONTENT"
            gate["severity"] = "fatal"
    decision = decide(state["run_id"], state["policy_run_id"], 1, bundle, flipped)
    assert decision["decision_action"] == "REJECT_CONTENT"
    assert decision["search_query_effect_status"] == "rule_blocked"


def test_judge_failed_technical_retry_is_config_driven_not_code_special_case(tmp_path):
    # Gemini technical failure is config-driven, not a code special-case.
    state = _state(tmp_path)
    bundle = deepcopy(state["evidence_bundles"][0])
    bundle["pattern_match_result"]["judge_status"] = "failed"

    decision = decide(state["run_id"], state["policy_run_id"], 1, bundle, state["policy_bundle"])

    assert decision["decision_action"] == "TECHNICAL_RETRY_REQUIRED"
    assert decision["decision_reason_code"] == "v4_technical_retry_needed"
    assert decision["search_query_effect_status"] == "failed"
    assert decision["triggered_blocking_rules"] == ["judge_failed"]
    assert decision["score"] is None
    assert decision["decision_replay_data"]["effect_mapping_id"] == "map_technical_retry_failed_hard_gate"


def test_legacy_low_confidence_field_no_longer_triggers_hard_gate(tmp_path):
    state = _state(tmp_path)
    bundle = deepcopy(state["evidence_bundles"][0])
    bundle["pattern_match_result"]["fit_confidence"] = 0.4

    decision = decide(state["run_id"], state["policy_run_id"], 1, bundle, state["policy_bundle"])

    assert "low_confidence" not in decision["triggered_blocking_rules"]
    assert decision["decision_reason_code"] != "content_low_confidence"
    assert decision["decision_reason_code"] != "age_50_plus_weak"


def test_low_confidence_lt_boundary_passes_at_threshold(tmp_path):
    # lt 0.6 boundary: exactly 0.6 is NOT low confidence, so the gate does not fire and
    # the content falls through to scoring instead of being rule-blocked.
    state = _state(tmp_path)
    bundle = deepcopy(state["evidence_bundles"][0])
    bundle["pattern_match_result"]["fit_confidence"] = 0.6

    decision = decide(state["run_id"], state["policy_run_id"], 1, bundle, state["policy_bundle"])

    assert "low_confidence" not in decision["triggered_blocking_rules"]
    assert decision["decision_reason_code"] != "content_low_confidence"
    assert decision["search_query_effect_status"] != "rule_blocked"


def _state(tmp_path):
    service = RunService(
        runtime_root=tmp_path / "runtime" / "v1",
        query_variant_client=FakeQueryVariantClient(),
    )
    return service.start_run(
        RunStartRequest(platform_mode="mock", source=str(REAL_SOURCE_FIXTURE))
    )