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

from __future__ import annotations

import copy
from typing import Any, Callable

from content_agent.business_modules.run_record.validation import validate_run
from content_agent.integrations.runtime_files import LocalRuntimeFileStore, RUNTIME_FILENAMES


RUN_ID = "run_v4_contract"
POLICY_RUN_ID = "policy_v4_contract"


def test_v4_contract_runtime_passes_validate_run(tmp_path):
    runtime = _write_runtime(tmp_path)

    result = validate_run(RUN_ID, runtime)

    assert result["status"] == "pass"


def test_v4_contract_does_not_apply_to_v3_scorecard_records(tmp_path):
    def mutate(data: dict[str, Any]) -> None:
        decision = data["rule_decisions.jsonl"][0]
        decision["score"] = None
        decision["scorecard"] = {
            "total_score": None,
            "fit_senior_50plus": True,
            "relevance_score": 0.91,
        }
        decision["decision_replay_data"].pop("allow_walk")

    runtime = _write_runtime(tmp_path, mutate)

    result = validate_run(RUN_ID, runtime)

    assert result["status"] == "pass"


def test_v4_score_contract_rejects_bad_total(tmp_path):
    def mutate(data: dict[str, Any]) -> None:
        data["rule_decisions.jsonl"][0]["score"] = 92
        _refresh_final_output_explanations(data)

    runtime = _write_runtime(tmp_path, mutate)

    result = validate_run(RUN_ID, runtime)

    assert _check_ids(result) == ["v4_score_total_mismatch"]


def test_v4_walk_gate_rejects_allow_walk_below_threshold(tmp_path):
    def mutate(data: dict[str, Any]) -> None:
        decision = data["rule_decisions.jsonl"][0]
        decision["score"] = 65
        decision["scorecard"]["query_relevance_score"] = 60
        decision["scorecard"]["platform_performance_score"] = 70
        _refresh_final_output_explanations(data)

    runtime = _write_runtime(tmp_path, mutate)

    result = validate_run(RUN_ID, runtime)

    assert "v4_allow_walk_threshold_mismatch" in _check_ids(result)


def test_v4_walk_gate_rejects_allow_walk_false_at_passing_threshold(tmp_path):
    def mutate(data: dict[str, Any]) -> None:
        data["rule_decisions.jsonl"][0]["decision_replay_data"]["allow_walk"] = False
        _refresh_final_output_explanations(data)

    runtime = _write_runtime(tmp_path, mutate)

    result = validate_run(RUN_ID, runtime)

    assert "v4_allow_walk_threshold_mismatch" in _check_ids(result)


def test_v4_walk_action_rejects_success_when_allow_walk_false(tmp_path):
    def mutate(data: dict[str, Any]) -> None:
        decision = data["rule_decisions.jsonl"][0]
        decision["score"] = 70
        decision["scorecard"]["platform_performance_score"] = 60
        decision["decision_replay_data"]["allow_walk"] = False
        decision["decision_replay_data"]["allow_walk_reason"] = "v4_query_and_platform_pass"
        decision["decision_replay_data"]["walk_gate_snapshot"] = {
            "query_relevance_score": 80,
            "platform_performance_score": 60,
            "score": 70,
        }
        _refresh_final_output_explanations(data)
        data["walk_actions.jsonl"].append(
            _walk_action(
                "walk_page_001",
                "hashtag_to_query",
                "success",
                {
                    "decision_id": "decision_001",
                    "allow_walk": False,
                    "allow_walk_reason": "v4_query_and_platform_pass",
                    "walk_gate_snapshot": decision["decision_replay_data"]["walk_gate_snapshot"],
                },
            )
        )

    runtime = _write_runtime(tmp_path, mutate)

    result = validate_run(RUN_ID, runtime)

    assert "v4_walk_action_allow_walk_denied" in _check_ids(result)


def test_v4_walk_action_rejects_missing_gate_snapshot(tmp_path):
    def mutate(data: dict[str, Any]) -> None:
        data["walk_actions.jsonl"].append(
            _walk_action(
                "walk_tag_001",
                "hashtag_to_query",
                "success",
                {
                    "decision_id": "decision_001",
                    "allow_walk": True,
                    "allow_walk_reason": "query>=70/platform>=65/score>=70",
                },
            )
        )

    runtime = _write_runtime(tmp_path, mutate)

    result = validate_run(RUN_ID, runtime)

    assert "v4_walk_action_gate_context_missing" in _check_ids(result)


def test_v4_action_thresholds_reject_conflicting_action(tmp_path):
    def mutate(data: dict[str, Any]) -> None:
        decision = data["rule_decisions.jsonl"][0]
        decision["decision_action"] = "KEEP_CONTENT_FOR_REVIEW"

    runtime = _write_runtime(tmp_path, mutate)

    result = validate_run(RUN_ID, runtime)

    assert "v4_action_threshold_mismatch" in _check_ids(result)


def test_v4_gemini_failure_requires_structured_failure_fields(tmp_path):
    def mutate(data: dict[str, Any]) -> None:
        summary = data["pattern_recall_evidence.jsonl"][0]["evidence_summary"]
        summary.clear()
        summary.update(
            {
                "schema_version": "v4_gemini_query_relevance.v1",
                "final_status": "failed",
                "retry_count": 0,
            }
        )

    runtime = _write_runtime(tmp_path, mutate)

    result = validate_run(RUN_ID, runtime)

    assert {
        "v4_gemini_failure_incomplete",
        "v4_gemini_failure_retry_invalid",
    } <= set(_check_ids(result))


def test_v4_legacy_field_blocklist_rejects_v4_records_only(tmp_path):
    def mutate(data: dict[str, Any]) -> None:
        data["rule_decisions.jsonl"][0]["scorecard"]["platform_heat"] = 88
        data["pattern_recall_evidence.jsonl"][0]["evidence_summary"]["relevance_score"] = 0.9

    runtime = _write_runtime(tmp_path, mutate)

    result = validate_run(RUN_ID, runtime)

    assert _check_ids(result).count("v4_legacy_field_present") == 2


def test_v4_final_output_requires_explanation(tmp_path):
    def mutate(data: dict[str, Any]) -> None:
        data["final_output.json"]["decision_records"][0].pop("v4_explanation")

    runtime = _write_runtime(tmp_path, mutate)

    result = validate_run(RUN_ID, runtime)

    assert "v4_final_output_explanation_missing" in _check_ids(result)


def test_v4_final_output_explanation_must_match_decision(tmp_path):
    def mutate(data: dict[str, Any]) -> None:
        data["final_output.json"]["decision_records"][0]["v4_explanation"]["allow_walk"] = False

    runtime = _write_runtime(tmp_path, mutate)

    result = validate_run(RUN_ID, runtime)

    assert "v4_final_output_explanation_mismatch" in _check_ids(result)


def test_v4_strategy_review_requires_summary_when_generated(tmp_path):
    def mutate(data: dict[str, Any]) -> None:
        data["strategy_review.json"].pop("v4_summary")

    runtime = _write_runtime(tmp_path, mutate)

    result = validate_run(RUN_ID, runtime)

    assert "v4_strategy_review_explanation_missing" in _check_ids(result)


def _write_runtime(
    tmp_path,
    mutate: Callable[[dict[str, Any]], None] | None = None,
) -> LocalRuntimeFileStore:
    runtime = LocalRuntimeFileStore(tmp_path / "runtime")
    runtime.prepare_run(RUN_ID)
    data = _runtime_payload()
    if mutate:
        mutate(data)
    for filename in RUNTIME_FILENAMES:
        value = data[filename]
        if isinstance(value, list):
            runtime.append_jsonl(RUN_ID, filename, value)
        else:
            runtime.write_json(RUN_ID, filename, value)
    return runtime


def _runtime_payload() -> dict[str, Any]:
    evidence_pack = {
        "pattern_source_system": "pg_pattern_v2",
        "case_id_type": "post_id",
        "source_kind": "pattern_itemset",
        "source_post_id": "post_001",
        "pattern_execution_id": 581,
        "mining_config_id": 2082,
        "itemset_ids": [1608352],
        "itemset_items": ["毛主席", "感人"],
        "support": 5,
        "absolute_support": 5,
        "matched_post_ids": ["post_001", "post_002"],
        "video_ids": ["video_001"],
        "case_ids": ["case_001"],
        "seed_terms": ["父爱感悟"],
        "discovery_start_source": "pattern_seed",
        "previous_discovery_step": "search_query_generated",
        "origin_path_id": "path_origin_001",
        "run_id": RUN_ID,
        "policy_run_id": POLICY_RUN_ID,
        "source_certainty": "high",
        "validation_status": "validated",
    }
    source_evidence = _source_evidence(evidence_pack, "content_001")
    path_ids = ["path_pattern_query", "path_query_content", "path_decision_asset"]
    decision = {
        "record_schema_version": "runtime_record.v1",
        "run_id": RUN_ID,
        "policy_run_id": POLICY_RUN_ID,
        "decision_id": "decision_001",
        "policy_bundle_id": "policy_bundle_v4",
        "rule_pack_id": "douyin_content_discovery_rule_pack_v4",
        "rule_pack_version": "4.0.0",
        "strategy_version": "V4",
        "decision_target_type": "content",
        "decision_target_id": "content_001",
        "decision_action": "ADD_TO_CONTENT_POOL",
        "decision_reason_code": "v4_query_and_platform_pass",
        "search_query_effect_status": "success",
        "score": 75,
        "scorecard": {
            "schema_version": "v4_scorecard.v1",
            "query_relevance_score": 80,
            "platform_performance_score": 70,
            "missing_observable_fields": [],
        },
        "source_evidence": copy.deepcopy(source_evidence),
        "decision_replay_data": {
            "policy_bundle_hash": "hash_v4",
            "rule_pack_id": "douyin_content_discovery_rule_pack_v4",
            "rule_pack_version": "4.0.0",
            "dispatch_id": "dispatch_v4",
            "strategy_version": "V4",
            "allow_walk": True,
            "allow_walk_reason": "query>=70/platform>=65/score>=70",
            "walk_gate_snapshot": {
                "query_relevance_score": 80,
                "platform_performance_score": 70,
                "score": 75,
            },
        },
        "raw_payload": {"decision_id": "decision_001", "v4_contract": True},
    }
    v4_explanation = _v4_explanation(decision)
    return {
        "source_context.json": {
            "schema_version": "runtime_record.v1",
            "run_id": RUN_ID,
            "demand_content_id": "123",
            "ext_data": {"evidence_pack": evidence_pack},
        },
        "pattern_seed_pack.json": {
            "schema_version": "runtime_record.v1",
            "run_id": RUN_ID,
            "policy_run_id": POLICY_RUN_ID,
            "itemsets": [{"itemset_id": 1608352}],
            "seed_terms": ["父爱感悟"],
        },
        "search_queries.jsonl": [
            {
                "record_schema_version": "runtime_record.v1",
                "run_id": RUN_ID,
                "policy_run_id": POLICY_RUN_ID,
                "search_query_id": "query_001",
                "search_query": "父爱感悟",
                "search_query_generation_method": "v4_seed",
                "discovery_start_source": "pattern_seed",
                "previous_discovery_step": "search_query_generated",
                "search_query_effect_status": "success",
                "pattern_seed_ref": {"query_source_type": "seed"},
                "raw_payload": {"query_source_refs": [{"query_source_type": "seed"}]},
            }
        ],
        "discovered_content_items.jsonl": [
            {
                "record_schema_version": "runtime_record.v1",
                "run_id": RUN_ID,
                "policy_run_id": POLICY_RUN_ID,
                "content_discovery_id": "discovery_001",
                "search_query_id": "query_001",
                "platform": "douyin",
                "platform_content_id": "content_001",
                "content_url": "https://example.com/content_001",
                "statistics": {"share_count": 10},
                "tags": ["父爱"],
                "source_evidence": copy.deepcopy(source_evidence),
                "pattern_match_result": {
                    "judge_status": "ok",
                    "pattern_recall_evidence_id": "recall_001",
                },
                "platform_raw_payload": {},
                "raw_payload": {"platform_content_id": "content_001"},
            }
        ],
        "content_media_records.jsonl": [
            {
                "record_schema_version": "runtime_record.v1",
                "run_id": RUN_ID,
                "policy_run_id": POLICY_RUN_ID,
                "media_record_id": "media_001",
                "platform": "douyin",
                "platform_content_id": "content_001",
                "media_status": "available",
                "raw_payload": {"platform_content_id": "content_001"},
            }
        ],
        "pattern_recall_evidence.jsonl": [
            {
                "record_schema_version": "runtime_record.v1",
                "run_id": RUN_ID,
                "policy_run_id": POLICY_RUN_ID,
                "recall_evidence_id": "recall_001",
                "content_discovery_id": "discovery_001",
                "platform_content_id": "content_001",
                "recall_status": "judged",
                "evidence_summary": {
                    "schema_version": "v4_gemini_query_relevance.v1",
                    "final_status": "success",
                    "query_relevance_score": 80,
                    "reason": "契合 query",
                },
                "raw_payload": {"recall_evidence_id": "recall_001"},
            }
        ],
        "rule_decisions.jsonl": [decision],
        "walk_actions.jsonl": [],
        "run_events.jsonl": [],
        "source_path_records.jsonl": [
            _path("path_pattern_query", "pattern_to_search_query", "Pattern", 581, "SearchQuery", "query_001"),
            _path("path_query_content", "search_query_to_content", "SearchQuery", "query_001", "Content", "content_001"),
            _path(
                "path_decision_asset",
                "decision_to_asset",
                "RuleDecision",
                "decision_001",
                "ContentAsset",
                "content_001",
                decision_id="decision_001",
            ),
        ],
        "search_clues.jsonl": [
            {
                "record_schema_version": "runtime_record.v1",
                "run_id": RUN_ID,
                "policy_run_id": POLICY_RUN_ID,
                "clue_id": "clue_001",
                "search_query_id": "query_001",
                "search_query": "父爱感悟",
                "result_count": 1,
                "pooled_content_count": 1,
                "review_content_count": 0,
                "pending_content_count": 0,
                "rejected_content_count": 0,
                "search_query_effect_status": "success",
                "query_aggregation_id": "agg_query_success",
                "raw_payload": {"clue_id": "clue_001"},
            }
        ],
        "final_output.json": {
            "schema_version": "runtime_record.v1",
            "run_id": RUN_ID,
            "policy_run_id": POLICY_RUN_ID,
            "validation_status": "pass",
            "content_assets": [
                {
                    "platform_content_id": "content_001",
                    "decision_id": "decision_001",
                    "source_path_record_ids": path_ids,
                    "source_evidence": copy.deepcopy(source_evidence),
                    "v4_explanation": copy.deepcopy(v4_explanation),
                }
            ],
            "author_assets": [],
            "review_records": [],
            "decision_records": [
                {
                    "decision_id": "decision_001",
                    "score": 75,
                    "scorecard": copy.deepcopy(decision["scorecard"]),
                    "decision_replay_data": copy.deepcopy(decision["decision_replay_data"]),
                    "source_evidence": copy.deepcopy(source_evidence),
                    "v4_explanation": copy.deepcopy(v4_explanation),
                }
            ],
            "search_clues": [],
            "reject_records": [],
            "summary": {
                "pooled_content_count": 1,
                "review_content_count": 0,
                "pending_content_count": 0,
                "rejected_content_count": 0,
                "run_path_complete": True,
                "trace_complete": True,
            },
        },
        "strategy_review.json": {
            "schema_version": "runtime_record.v1",
            "run_id": RUN_ID,
            "policy_run_id": POLICY_RUN_ID,
            "summary": {},
            "v4_summary": {
                "schema_version": "v4_strategy_review_summary.v1",
                "score_buckets": {
                    "pool": 1,
                    "review": 0,
                    "reject": 0,
                    "technical_retry": 0,
                    "unknown": 0,
                },
                "allow_walk_distribution": {
                    "allowed": 1,
                    "denied": 0,
                    "missing": 0,
                },
                "walk_gate_review": {
                    "v4_gate_distribution": {
                        "allowed": 0,
                        "denied": 0,
                        "missing": 0,
                    }
                },
            },
            "raw_payload": {"strategy_review_id": "review_001"},
        },
    }


def _v4_explanation(decision: dict[str, Any]) -> dict[str, Any]:
    scorecard = decision["scorecard"]
    replay = decision["decision_replay_data"]
    return {
        "schema_version": "v4_decision_explanation.v1",
        "scorecard_schema_version": scorecard["schema_version"],
        "query_relevance_score": scorecard["query_relevance_score"],
        "platform_performance_score": scorecard["platform_performance_score"],
        "score": decision["score"],
        "platform_performance_components": scorecard.get("platform_performance_components", []),
        "missing_observable_fields": scorecard.get("missing_observable_fields", []),
        "decision_action": decision["decision_action"],
        "decision_reason_code": decision["decision_reason_code"],
        "search_query_effect_status": decision["search_query_effect_status"],
        "allow_walk": replay["allow_walk"],
        "allow_walk_reason": replay["allow_walk_reason"],
        "walk_gate_snapshot": replay["walk_gate_snapshot"],
    }


def _refresh_final_output_explanations(data: dict[str, Any]) -> None:
    decisions = {decision["decision_id"]: decision for decision in data["rule_decisions.jsonl"]}
    for section in ["content_assets", "review_records", "reject_records", "decision_records"]:
        for record in data["final_output.json"].get(section, []):
            decision = decisions.get(record.get("decision_id"))
            if not decision:
                continue
            record["v4_explanation"] = _v4_explanation(decision)
            if section == "decision_records":
                record["score"] = decision.get("score")
                record["scorecard"] = copy.deepcopy(decision.get("scorecard"))
                record["decision_replay_data"] = copy.deepcopy(
                    decision.get("decision_replay_data")
                )


def _source_evidence(evidence_pack: dict[str, Any], platform_content_id: str) -> dict[str, Any]:
    evidence = copy.deepcopy(evidence_pack)
    evidence["discovered_platform_content_id"] = platform_content_id
    return evidence


def _path(
    path_id: str,
    path_type: str,
    from_type: str,
    from_id: Any,
    to_type: str,
    to_id: Any,
    decision_id: str | None = None,
) -> dict[str, Any]:
    return {
        "record_schema_version": "runtime_record.v1",
        "run_id": RUN_ID,
        "policy_run_id": POLICY_RUN_ID,
        "source_path_record_id": path_id,
        "source_path_type": path_type,
        "from_node_type": from_type,
        "from_node_id": from_id,
        "to_node_type": to_type,
        "to_node_id": to_id,
        "decision_id": decision_id,
        "raw_payload": {"source_path_record_id": path_id},
    }


def _walk_action(
    action_id: str,
    edge_id: str,
    status: str,
    raw_payload: dict[str, Any],
) -> dict[str, Any]:
    return {
        "record_schema_version": "runtime_record.v1",
        "run_id": RUN_ID,
        "policy_run_id": POLICY_RUN_ID,
        "walk_action_id": action_id,
        "edge_id": edge_id,
        "edge_type": "query",
        "from_node_type": "SearchQuery",
        "from_node_id": "query_001",
        "to_node_type": "SearchQuery",
        "to_node_id": "query_002",
        "walk_action": "fetch_next_page",
        "walk_status": status,
        "budget_tier": "normal",
        "depth": 1,
        "reason_code": None,
        "raw_payload": raw_payload,
    }


def _check_ids(result: dict[str, Any]) -> list[str]:
    return [finding["check_id"] for finding in result["findings"]]