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 _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": {},
    }