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

from copy import deepcopy
from pathlib import Path

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


def test_rule_pack_thresholds_drive_decision(tmp_path):
    service = RunService(
        runtime_root=tmp_path / "runtime" / "v1",
        query_variant_client=FakeQueryVariantClient(),
    )
    state = service.start_run(
        RunStartRequest(platform_mode="mock", source=str(REAL_SOURCE_FIXTURE))
    )
    run_id = state["run_id"]

    bundle = deepcopy(state["evidence_bundles"][0])
    bundle["pattern_match_result"]["query_relevance_score"] = 80
    bundle["content_engagement_metrics"]["platform_performance"]["platform_performance_score"] = 60

    decision = decide(run_id, state["policy_run_id"], 1, bundle, state["policy_bundle"])
    assert decision["score"] == 70
    assert decision["decision_action"] == "ADD_TO_CONTENT_POOL"
    assert decision["decision_reason_code"] == "v4_query_and_platform_pass"
    assert decision["decision_replay_data"]["allow_walk"] is False
    assert decision["policy_run_id"] == state["policy_run_id"]
    assert decision["strategy_version"] == "V4"


def test_rule_pack_hard_gate_reason_code_drives_decision(tmp_path):
    service = RunService(
        runtime_root=tmp_path / "runtime" / "v1",
        query_variant_client=FakeQueryVariantClient(),
    )
    state = service.start_run(
        RunStartRequest(platform_mode="mock", source=str(REAL_SOURCE_FIXTURE))
    )
    run_id = state["run_id"]

    bundle = deepcopy(state["evidence_bundles"][0])
    bundle["source_evidence"] = {}

    decision = decide(run_id, state["policy_run_id"], 1, bundle, state["policy_bundle"])
    assert decision["decision_action"] == "REJECT_CONTENT"
    assert decision["decision_reason_code"] == "missing_source_evidence"
    assert decision["search_query_effect_status"] == "rule_blocked"
    assert decision["triggered_blocking_rules"] == ["missing_source_evidence"]
    assert decision["decision_replay_data"]["primary_reason_code"] == "missing_source_evidence"


def test_missing_score_uses_rule_pack_missing_policy(tmp_path):
    service = RunService(
        runtime_root=tmp_path / "runtime" / "v1",
        query_variant_client=FakeQueryVariantClient(),
    )
    state = service.start_run(
        RunStartRequest(platform_mode="mock", source=str(REAL_SOURCE_FIXTURE))
    )
    run_id = state["run_id"]

    bundle = deepcopy(state["evidence_bundles"][0])
    bundle["pattern_match_result"]["query_relevance_score"] = None
    bundle["content_engagement_metrics"]["platform_performance"]["platform_performance_score"] = None

    decision = decide(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["scorecard"]["score_missing"] is True


def test_unknown_strategy_version_fails_before_rule_decision(tmp_path):
    service = RunService(
        runtime_root=tmp_path / "runtime" / "v1",
        query_variant_client=FakeQueryVariantClient(),
    )

    state = service.start_run(
        RunStartRequest(
            platform_mode="mock",
            strategy_version="missing_strategy",
            source=str(REAL_SOURCE_FIXTURE),
        )
    )

    assert state["status"] == "failed"
    assert state["error_code"] == ErrorCode.POLICY_BUNDLE_NOT_FOUND.value
    assert state["errors"] == ["policy bundle not found"]


def test_rule_judgment_does_not_call_pattern_recall_integrations():
    root = Path("content_agent/business_modules/rule_judgment")
    text = "\n".join(path.read_text(encoding="utf-8") for path in root.rglob("*.py"))

    assert "decode_api" not in text
    assert "category_match" not in text
    assert "match-paths" not in text


def test_disabled_future_packs_not_marked_as_active_entity_packs():
    from content_agent.integrations.policy_json import JsonPolicyBundleStore

    bundle = JsonPolicyBundleStore().load_policy_bundle("V4")

    assert set(bundle["rule_pack_by_entity"]) == {"Content"}
    for entity in ("Author", "Hashtag", "Path", "Budget"):
        assert entity not in bundle["rule_pack_by_entity"]