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

"""config x case matrix (V2-M0E).

Replays the same captured case under different configurations to prove the
"foolproof config" safety net: changing config changes the case outcome,
visibly (snapshot diff), without breaking the pipeline. Variants that depend on
later modules (M3 per-entity dispatch) are xfail until then.
"""

from __future__ import annotations

import json
import shutil
from pathlib import Path

import pytest

from content_agent.integrations.policy_json import JsonPolicyBundleStore
from tests.replay_harness import replay_case
from tests.snapshot import assert_matches

ROOT = Path(__file__).resolve().parents[1]
_RULE_PACK_REL = "product_documents/规则包/douyin_rule_packs.v1.json"
_WALK_REL = "product_documents/抖音游走策略/douyin_walk_strategy.v1.json"


def _senior_block_store(root: Path) -> JsonPolicyBundleStore:
    """M3 config variant: flip the not_fit_senior gate to fire on fit_senior_50plus == true.

    The captured case's mock Gemini judgment marks every item fit (fit_senior_50plus=true),
    so inverting the gate's expected value blocks the whole batch by config alone — a clean
    counterproof that the hard gate (and the downstream walk) is config-driven, not hardcoded.
    """
    (root / _RULE_PACK_REL).parent.mkdir(parents=True, exist_ok=True)
    (root / _WALK_REL).parent.mkdir(parents=True, exist_ok=True)
    shutil.copy(ROOT / _WALK_REL, root / _WALK_REL)
    package = json.loads((ROOT / _RULE_PACK_REL).read_text(encoding="utf-8"))
    for pack in package.get("rule_packs", []):
        for gate in pack.get("hard_gates", []):
            if gate.get("gate_id") == "not_fit_senior":
                gate["when"]["value"] = True
    (root / _RULE_PACK_REL).write_text(json.dumps(package, ensure_ascii=False, indent=2), encoding="utf-8")
    return JsonPolicyBundleStore(root)


def _outcome(artifacts) -> dict:
    return {
        "reasons": sorted(d.get("decision_reason_code") for d in artifacts.decisions),
        "effect_status_counts": artifacts.summary.get("effect_status_counts"),
        "pooled": artifacts.summary.get("pooled_content_count"),
        "rejected": artifacts.summary.get("rejected_content_count"),
    }


def _variant_overrides(variant: str, cfg_dir: Path):
    if variant == "default":
        return None
    if variant == "senior_block":
        return {"policy_store": _senior_block_store(cfg_dir)}
    raise ValueError(variant)


@pytest.mark.parametrize("variant", ["default", "senior_block"])
def test_matrix_real_id45(variant, tmp_path):
    overrides = _variant_overrides(variant, tmp_path / "cfg")
    artifacts = replay_case("real_id45", runtime_root=tmp_path / "rt", config_overrides=overrides)
    assert artifacts.state["status"] == "success"  # config change must not break the chain
    assert_matches(f"matrix/real_id45__{variant}", _outcome(artifacts))


def test_senior_block_changes_outcome(tmp_path):
    base = _outcome(replay_case("real_id45", runtime_root=tmp_path / "rt0"))
    blocked = _outcome(
        replay_case(
            "real_id45",
            runtime_root=tmp_path / "rt1",
            config_overrides={"policy_store": _senior_block_store(tmp_path / "cfg")},
        )
    )
    # Decoupling proof: one config edit on the not_fit_senior gate visibly moves the outcome.
    assert base != blocked
    # Default: no item is blocked by the senior-fit gate; items flow into pool / review.
    assert "content_not_fit_senior" not in base["reasons"]
    assert base["effect_status_counts"]["rule_blocked"] == 0
    # R3 第二步: 四字段热度复合后 real_id45 默认 3 进池(原 2)。
    assert base["pooled"] == 3
    # Blocked variant: every item trips the (config-inverted) hard gate -> rule_blocked reject.
    assert blocked["reasons"] == ["content_not_fit_senior"] * 4
    assert blocked["effect_status_counts"]["rule_blocked"] == 4
    assert blocked["pooled"] == 0
    assert blocked["rejected"] == 4


def test_matrix_query_profile_variant():
    from scripts.validate_query_prompts_config import validate_query_prompts_config

    config = json.loads((ROOT / "product_documents/配置/query_prompts.v1.json").read_text(encoding="utf-8"))
    assert validate_query_prompts_config(config) == []


def test_decoupling_counterproof():
    # M3A removed the Content hardcode: dispatch is parametrized by target_entity,
    # so a non-Content (e.g. Author) pack can be routed without falling back.
    source = (ROOT / "content_agent/integrations/policy_json.py").read_text(encoding="utf-8")
    assert 'target_entity") == "Content"' not in source


def test_senior_block_blocks_all_walk_expansion(tmp_path):
    # M4 受控变化: 全拦截(rule_blocked)时翻页/作者/tag 全停;砍包后 path_stop 戳=内容包。
    artifacts = replay_case(
        "real_id45",
        runtime_root=tmp_path / "rt",
        config_overrides={"policy_store": _senior_block_store(tmp_path / "cfg")},
    )
    walk_actions = artifacts.files["walk_actions.jsonl"]

    assert not [row for row in walk_actions if row["edge_id"] == "query_next_page"]
    expansions = [
        row for row in walk_actions if row["edge_id"] in {"author_to_works", "hashtag_to_query"}
    ]
    assert expansions
    assert all(row["walk_status"] == "skipped" for row in expansions)
    assert all(row["reason_code"] == "blocked_by_rule_decision" for row in expansions)

    path_stops = [row for row in walk_actions if row["edge_id"] == "path_stop"]
    assert len(path_stops) == 4
    for row in path_stops:
        assert row["rule_pack_id"] == "douyin_content_discovery_rule_pack_v1"
        assert row["raw_payload"]["rule_pack_execution"]["executed_rule_pack_id"] == (
            "douyin_content_discovery_rule_pack_v1"
        )