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

from content_agent.business_modules import learning_review
from content_agent.integrations.runtime_files import LocalRuntimeFileStore


def test_strategy_review_outputs_p8_contract_with_missing_feedback(tmp_path):
    runtime = LocalRuntimeFileStore(tmp_path / "runtime")
    run_id = "run_001"
    policy_run_id = "policy_001"
    runtime.prepare_run(run_id)
    _write_minimal_runtime(runtime, run_id, policy_run_id)

    review = learning_review.run(run_id, policy_run_id, runtime)

    assert review["data_window"]["scope"] == "single_run"
    assert review["metric_summary"]["search_query_count"] == 4
    assert review["metric_summary"]["pooled_content_count"] == 1
    assert review["query_review"]["effective_queries"][0]["search_query"] == "银发旅行"
    assert review["query_review"]["review_queries"][0]["search_query_effect_status"] == "pending"
    assert review["rule_review"]["decision_distribution"]["ADD_TO_CONTENT_POOL"] == 1
    assert review["walk_review"]["walk_action_count"] == 1
    assert review["asset_review"]["content_asset_count"] == 1
    assert review["asset_review"]["search_clue_assets"]["promoted_count"] == 1
    assert review["performance_feedback"]["performance_feedback_status"] == "missing"
    assert review["recommendations"]
    assert all(item["requires_human_approval"] is True for item in review["recommendations"])

    written = runtime.read_json(run_id, "strategy_review.json")
    assert written["raw_payload"]["recommendations"] == review["recommendations"]


def test_strategy_review_filters_legacy_contract_values(tmp_path):
    runtime = LocalRuntimeFileStore(tmp_path / "runtime")
    run_id = "run_legacy"
    policy_run_id = "policy_legacy"
    runtime.prepare_run(run_id)
    _write_minimal_runtime(runtime, run_id, policy_run_id)
    runtime.append_jsonl(
        run_id,
        "search_clues.jsonl",
        [
            {
                "run_id": run_id,
                "policy_run_id": policy_run_id,
                "clue_id": "clue_old",
                "search_query_id": "q_old",
                "search_query": "旧口径",
                "search_query_effect_status": "weak_" + "effective",
                "result_count": 10,
                "pooled_content_count": 10,
                "review_content_count": 0,
                "pending_content_count": 0,
                "rejected_content_count": 0,
                "raw_payload": {},
            }
        ],
    )
    runtime.append_jsonl(
        run_id,
        "rule_decisions.jsonl",
        [
            {
                "run_id": run_id,
                "policy_run_id": policy_run_id,
                "decision_id": "decision_old",
                "decision_target_type": "content",
                "decision_target_id": "content_old",
                "decision_action": "HOLD_CONTENT_" + "PENDING",
                "decision_reason_code": "old_status",
                "search_query_effect_status": "block" + "ed",
                "raw_payload": {},
            }
        ],
    )

    review = learning_review.run(run_id, policy_run_id, runtime)

    assert "weak_" + "effective" not in {
        item["search_query_effect_status"]
        for bucket in review["query_review"].values()
        if isinstance(bucket, list)
        for item in bucket
    }
    assert "HOLD_CONTENT_" + "PENDING" not in review["rule_review"]["decision_distribution"]


def test_strategy_review_builds_v4_single_run_summary(tmp_path):
    runtime = LocalRuntimeFileStore(tmp_path / "runtime")
    run_id = "run_v4"
    policy_run_id = "policy_v4"
    runtime.prepare_run(run_id)
    _write_v4_runtime(runtime, run_id, policy_run_id)

    review = learning_review.run(run_id, policy_run_id, runtime)

    v4_summary = review["v4_summary"]
    assert v4_summary["schema_version"] == "v4_strategy_review_summary.v1"
    assert v4_summary["summary_status"] == "available"
    assert v4_summary["v4_decision_count"] == 4
    assert v4_summary["score_buckets"] == {
        "pool": 1,
        "review": 1,
        "reject": 1,
        "technical_retry": 1,
        "unknown": 0,
    }
    assert v4_summary["missing_observable_fields"] == [
        {"field": "statistics.play_count", "reason": "natural_platform_missing", "count": 1},
        {"field": "statistics.share_count", "reason": "runtime_missing", "count": 1},
    ]
    assert v4_summary["technical_retry"]["count"] == 1
    assert v4_summary["allow_walk_distribution"] == {
        "allowed": 1,
        "denied": 3,
        "missing": 0,
    }
    assert v4_summary["walk_gate_review"]["v4_gate_distribution"]["allowed"] == 1
    assert v4_summary["walk_gate_review"]["v4_gate_distribution"]["denied"] == 1
    assert v4_summary["walk_gate_review"]["deny_reasons"] == [
        {"reason_code": "v4_allow_walk_denied", "count": 1}
    ]
    assert review["walk_review"]["deny_reasons"] == [
        {"reason_code": "v4_allow_walk_denied", "count": 1}
    ]
    assert runtime.read_json(run_id, "strategy_review.json")["raw_payload"]["v4_summary"]


def _write_minimal_runtime(runtime, run_id: str, policy_run_id: str) -> None:
    runtime.write_json(
        run_id,
        "final_output.json",
        {
            "schema_version": "runtime_record.v1",
            "run_id": run_id,
            "policy_run_id": policy_run_id,
            "policy": {
                "strategy_version": "V1",
                "policy_bundle_id": "douyin_policy_bundle_v1",
                "rule_pack_id": "douyin_content_rules_v1",
                "rule_pack_version": "V1",
            },
            "walk_strategy": {"walk_strategy_version": "V1.0"},
            "content_assets": [
                {
                    "content_asset_id": "asset_001",
                    "platform": "douyin",
                    "platform_content_id": "content_001",
                    "source_path_record_ids": ["path_001"],
                    "decision_ids": ["decision_001"],
                }
            ],
            "author_assets": [{"author_asset_id": "author_001"}],
            "summary": {
                "search_query_count": 4,
                "discovered_content_count": 3,
                "pooled_content_count": 1,
                "review_content_count": 1,
                "rejected_content_count": 1,
            },
        },
    )
    runtime.append_jsonl(
        run_id,
        "search_clues.jsonl",
        [
            _clue(run_id, policy_run_id, "clue_001", "q_001", "银发旅行", "success", 1),
            _clue(run_id, policy_run_id, "clue_002", "q_002", "怀旧故事", "pending", 0),
            _clue(run_id, policy_run_id, "clue_003", "q_003", "广场舞", "failed", 0),
            _clue(run_id, policy_run_id, "clue_004", "q_004", "保健品", "rule_blocked", 0),
        ],
    )
    runtime.append_jsonl(
        run_id,
        "rule_decisions.jsonl",
        [
            _decision(
                run_id,
                policy_run_id,
                "decision_001",
                "content_001",
                "ADD_TO_CONTENT_POOL",
                "passed",
                "success",
                search_query_id="q_001",
            ),
            _decision(run_id, policy_run_id, "decision_002", "content_002", "KEEP_CONTENT_FOR_REVIEW", "needs_more_evidence", "pending"),
            _decision(run_id, policy_run_id, "decision_003", "content_003", "REJECT_CONTENT", "missing_content_portrait", "failed"),
        ],
    )
    runtime.append_jsonl(
        run_id,
        "discovered_content_items.jsonl",
        [
            {
                "run_id": run_id,
                "policy_run_id": policy_run_id,
                "content_discovery_id": "discovery_001",
                "search_query_id": "q_001",
                "platform": "douyin",
                "platform_content_id": "content_001",
                "raw_payload": {},
            }
        ],
    )
    runtime.append_jsonl(
        run_id,
        "source_path_records.jsonl",
        [
            {
                "run_id": run_id,
                "policy_run_id": policy_run_id,
                "source_path_record_id": "path_001",
                "source_path_type": "decision_to_asset",
                "from_node_type": "rule_decision",
                "from_node_id": "decision_001",
                "to_node_type": "content_asset",
                "to_node_id": "asset_001",
                "decision_id": "decision_001",
                "raw_payload": {},
            }
        ],
    )
    runtime.append_jsonl(
        run_id,
        "walk_actions.jsonl",
        [
            {
                "run_id": run_id,
                "policy_run_id": policy_run_id,
                "walk_action_id": "wa_001",
                "edge_id": "query_next_page",
                "walk_action": "fetch_next_page",
                "walk_status": "success",
                "raw_payload": {},
            }
        ],
    )


def _clue(run_id, policy_run_id, clue_id, query_id, query, status, pooled_count):
    return {
        "run_id": run_id,
        "policy_run_id": policy_run_id,
        "clue_id": clue_id,
        "search_query_id": query_id,
        "search_query": query,
        "search_query_effect_status": status,
        "result_count": 1,
        "pooled_content_count": pooled_count,
        "review_content_count": 1 if status == "pending" else 0,
        "pending_content_count": 1 if status == "pending" else 0,
        "rejected_content_count": 1 if status in {"failed", "rule_blocked"} else 0,
        "raw_payload": {},
    }


def _decision(
    run_id,
    policy_run_id,
    decision_id,
    target_id,
    action,
    reason,
    effect_status,
    search_query_id=None,
):
    return {
        "run_id": run_id,
        "policy_run_id": policy_run_id,
        "decision_id": decision_id,
        "decision_target_type": "content",
        "decision_target_id": target_id,
        "decision_action": action,
        "decision_reason_code": reason,
        "search_query_effect_status": effect_status,
        "source_evidence": {"search_query_id": search_query_id} if search_query_id else {},
        "raw_payload": {},
    }


def _write_v4_runtime(runtime, run_id: str, policy_run_id: str) -> None:
    runtime.write_json(
        run_id,
        "final_output.json",
        {
            "schema_version": "runtime_record.v1",
            "run_id": run_id,
            "policy_run_id": policy_run_id,
            "policy": {"strategy_version": "V4"},
            "walk_strategy": {"walk_strategy_version": "V4"},
            "content_assets": [{"content_asset_id": "asset_001"}],
            "author_assets": [],
            "summary": {
                "search_query_count": 4,
                "discovered_content_count": 4,
                "pooled_content_count": 1,
                "review_content_count": 2,
                "rejected_content_count": 1,
            },
        },
    )
    runtime.append_jsonl(
        run_id,
        "search_clues.jsonl",
        [
            _clue(run_id, policy_run_id, "clue_v4", "q_v4", "父爱感悟", "success", 1)
        ],
    )
    runtime.append_jsonl(
        run_id,
        "rule_decisions.jsonl",
        [
            _v4_decision(run_id, policy_run_id, "decision_pool", "content_pool", "ADD_TO_CONTENT_POOL", "success", 80, 70, True, []),
            _v4_decision(run_id, policy_run_id, "decision_review", "content_review", "KEEP_CONTENT_FOR_REVIEW", "pending", 60, 60, False, [{"field": "statistics.share_count", "reason": "runtime_missing"}]),
            _v4_decision(run_id, policy_run_id, "decision_reject", "content_reject", "REJECT_CONTENT", "failed", 40, 70, False, [{"field": "statistics.play_count", "reason": "natural_platform_missing"}]),
            _v4_decision(run_id, policy_run_id, "decision_retry", "content_retry", "KEEP_CONTENT_FOR_REVIEW", "pending", None, None, False, [], reason="v4_technical_retry_needed"),
        ],
    )
    runtime.append_jsonl(run_id, "discovered_content_items.jsonl", [])
    runtime.append_jsonl(run_id, "source_path_records.jsonl", [])
    runtime.append_jsonl(
        run_id,
        "walk_actions.jsonl",
        [
            {
                "run_id": run_id,
                "policy_run_id": policy_run_id,
                "walk_action_id": "walk_allowed",
                "edge_id": "hashtag_to_query",
                "walk_status": "success",
                "raw_payload": {
                    "decision_id": "decision_pool",
                    "allow_walk": True,
                    "allow_walk_reason": "query>=70/platform>=65/score>=70",
                    "walk_gate_status": "allowed",
                    "walk_gate_snapshot": {"query_relevance_score": 80, "platform_performance_score": 70, "score": 75},
                },
            },
            {
                "run_id": run_id,
                "policy_run_id": policy_run_id,
                "walk_action_id": "walk_denied",
                "edge_id": "author_to_works",
                "walk_status": "skipped",
                "reason_code": "v4_allow_walk_denied",
                "raw_payload": {
                    "decision_id": "decision_review",
                    "allow_walk": False,
                    "allow_walk_reason": "score below allow_walk threshold",
                    "walk_gate_status": "denied",
                    "walk_gate_reason_code": "v4_allow_walk_denied",
                    "walk_gate_snapshot": {"query_relevance_score": 60, "platform_performance_score": 60, "score": 60},
                },
            },
        ],
    )


def _v4_decision(
    run_id,
    policy_run_id,
    decision_id,
    target_id,
    action,
    effect_status,
    query_score,
    platform_score,
    allow_walk,
    missing_fields,
    reason=None,
):
    score = (
        round(query_score * 0.5 + platform_score * 0.5, 2)
        if query_score is not None and platform_score is not None
        else None
    )
    return {
        "run_id": run_id,
        "policy_run_id": policy_run_id,
        "decision_id": decision_id,
        "decision_target_type": "content",
        "decision_target_id": target_id,
        "decision_action": action,
        "decision_reason_code": reason or (
            "v4_query_and_platform_pass"
            if action == "ADD_TO_CONTENT_POOL"
            else "v4_score_review_needed"
            if action == "KEEP_CONTENT_FOR_REVIEW"
            else "v4_query_or_score_below_threshold"
        ),
        "search_query_effect_status": effect_status,
        "score": score,
        "scorecard": {
            "schema_version": "v4_scorecard.v1",
            "query_relevance_score": query_score,
            "platform_performance_score": platform_score,
            "missing_observable_fields": missing_fields,
            "score_missing": score is None,
            "failure_type": "llm_timeout" if score is None else None,
        },
        "decision_replay_data": {
            "allow_walk": allow_walk,
            "allow_walk_reason": (
                "query>=70/platform>=65/score>=70"
                if allow_walk
                else "score below allow_walk threshold"
            ),
            "walk_gate_snapshot": {
                "query_relevance_score": query_score,
                "platform_performance_score": platform_score,
                "score": score,
            },
        },
        "raw_payload": {},
    }