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_not_fit_senior_is_a_blocking_hard_gate(tmp_path):
    # M3: pattern_recall gate retired. The senior-fit judgment is now the blocking gate.
    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 decision["decision_action"] == "REJECT_CONTENT"
    assert decision["decision_reason_code"] == "content_not_fit_senior"
    assert decision["search_query_effect_status"] == "rule_blocked"
    assert decision["triggered_blocking_rules"] == ["not_fit_senior"]
    # Retired pattern-recall reason code must no longer surface.
    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 KEEP_CONTENT_FOR_REVIEW; 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"] == "KEEP_CONTENT_FOR_REVIEW"

    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_review_is_config_driven_not_code_special_case(tmp_path):
    # M3: the review/pending hard gate is now judge_failed (Gemini technical failure parks
    # content for human review). This 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"] == "KEEP_CONTENT_FOR_REVIEW"
    assert decision["decision_reason_code"] == "content_judge_failed"
    assert decision["search_query_effect_status"] == "pending"
    assert decision["triggered_blocking_rules"] == ["judge_failed"]
    assert decision["score"] is None
    assert decision["decision_replay_data"]["effect_mapping_id"] == "map_keep_for_review_pending_hard_gate"


def test_low_confidence_below_threshold_rejects(tmp_path):
    # M3: age_50_plus_weak gate retired. Low Gemini confidence is now the blocking gate
    # (fit_confidence lt 0.6 -> REJECT_CONTENT / content_low_confidence / rule_blocked).
    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 decision["decision_action"] == "REJECT_CONTENT"
    assert decision["decision_reason_code"] == "content_low_confidence"
    assert decision["search_query_effect_status"] == "rule_blocked"
    assert decision["triggered_blocking_rules"] == ["low_confidence"]
    # Retired age gate / reason code must no longer surface.
    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))
    )