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

"""config x case matrix.

Replays the same captured case under different configurations to prove the
"foolproof config" safety net: changing config changes the case outcome,
visibly, without breaking the pipeline.
"""

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

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"
_EDGE_CATALOG_REL = "tech_documents/数据接口与来源/walk_edge_catalog.json"


def _judge_ok_block_store(root: Path) -> JsonPolicyBundleStore:
    """Flip the existing judge_failed gate so judge_status == ok blocks by config alone."""
    (root / _RULE_PACK_REL).parent.mkdir(parents=True, exist_ok=True)
    (root / _WALK_REL).parent.mkdir(parents=True, exist_ok=True)
    (root / _EDGE_CATALOG_REL).parent.mkdir(parents=True, exist_ok=True)
    shutil.copy(ROOT / _WALK_REL, root / _WALK_REL)
    shutil.copy(ROOT / _EDGE_CATALOG_REL, root / _EDGE_CATALOG_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") == "judge_failed":
                gate["when"]["value"] = "ok"
                gate["decision_action"] = "REJECT_CONTENT"
                gate["severity"] = "fatal"
    (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 == "judge_ok_block":
        return {"policy_store": _judge_ok_block_store(cfg_dir)}
    raise ValueError(variant)


@pytest.mark.parametrize("variant", ["default", "judge_ok_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
    outcome = _outcome(artifacts)
    if variant == "default":
        # M11 re-baseline:3 入池(原第3条复看升入池)/ 1 淘汰。
        assert outcome["pooled"] == 3
        assert outcome["rejected"] == 1
        assert outcome["effect_status_counts"] == {
            "success": 3,
            "pending": 0,
            "failed": 1,
            "rule_blocked": 0,
        }
    else:
        assert outcome["pooled"] == 0
        assert outcome["rejected"] == 4
        assert outcome["effect_status_counts"]["rule_blocked"] == 4


def test_judge_ok_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": _judge_ok_block_store(tmp_path / "cfg")},
        )
    )
    assert base != blocked
    assert base["effect_status_counts"]["rule_blocked"] == 0
    assert base["pooled"] == 3
    assert blocked["reasons"] == ["v4_technical_retry_needed"] * 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_judge_ok_block_blocks_all_walk_expansion(tmp_path):
    artifacts = replay_case(
        "real_id45",
        runtime_root=tmp_path / "rt",
        config_overrides={"policy_store": _judge_ok_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"] == "v4_allow_walk_denied" for row in expansions)
    assert all(row["raw_payload"]["allow_walk"] is False 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"
        )