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

from content_agent.business_modules import learning_review
from content_agent.integrations.runtime_files import LocalRuntimeFileStore
from tests.test_p8_strategy_review import _write_minimal_runtime
from tests.test_v5_golden_corpus_baseline import _load_historical_decisions


class FeedbackRuntime(LocalRuntimeFileStore):
    def __init__(self, *args, feedback_rows=None, **kwargs):
        super().__init__(*args, **kwargs)
        self.feedback_rows = feedback_rows or []

    def read_performance_feedback(self, run_id: str, policy_run_id: str):
        return [
            row
            for row in self.feedback_rows
            if row["run_id"] == run_id and row["policy_run_id"] == policy_run_id
        ]


def test_strategy_review_summarizes_fake_performance_feedback(tmp_path):
    run_id = "run_feedback"
    policy_run_id = "policy_feedback"
    runtime = FeedbackRuntime(
        tmp_path / "runtime",
        feedback_rows=[
            {
                "run_id": run_id,
                "policy_run_id": policy_run_id,
                "feedback_id": "feedback_001",
                "feedback_status": "available",
                "completion_rate": 0.72,
                "share_rate": 0.08,
                "average_watch_seconds": 18.5,
            }
        ],
    )
    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["performance_feedback"]["performance_feedback_status"] == "available"
    assert review["performance_feedback"]["feedback_count"] == 1
    assert review["performance_feedback"]["average_completion_rate"] == 0.72
    assert any(
        item["recommendation_type"] == "performance_feedback"
        and item["suggested_action"] == "review_feedback_before_strategy_change"
        for item in review["recommendations"]
    )
    assert review["rule_review"]["decision_distribution"]["ADD_TO_CONTENT_POOL"] == 1


def _fake_feedback_for_decision(decision, platform="douyin"):
    platform_content_id = decision["decision_target_id"]
    return {
        "run_id": decision["run_id"],
        "policy_run_id": decision["policy_run_id"],
        "feedback_id": f"fake_m4_{decision['policy_run_id']}_{platform_content_id}",
        "feedback_source": "m4_fake_feedback",
        "feedback_status": "available",
        "platform": platform,
        "platform_content_id": platform_content_id,
        "completion_rate": 0.72,
    }


def _join_feedback(feedback_rows, discovered_rows, *, use_content_discovery_id=False):
    key_field = "content_discovery_id" if use_content_discovery_id else "platform_content_id"
    discovered_keys = {
        (row["run_id"], row["policy_run_id"], row.get(key_field))
        for row in discovered_rows
    }
    return [
        row
        for row in feedback_rows
        if (row["run_id"], row["policy_run_id"], row["platform_content_id"]) in discovered_keys
    ]


def test_fake_feedback_join_matches_m0_golden_samples():
    decisions = _load_historical_decisions("v4_douyin_57663")
    wanted_actions = {
        "ADD_TO_CONTENT_POOL",
        "KEEP_CONTENT_FOR_REVIEW",
        "REJECT_CONTENT",
        "TECHNICAL_RETRY_REQUIRED",
    }
    selected = []
    seen_actions = set()
    for decision in decisions:
        action = decision["decision_action"]
        if action in wanted_actions and action not in seen_actions:
            selected.append(decision)
            seen_actions.add(action)
        if seen_actions == wanted_actions:
            break

    feedback_rows = [_fake_feedback_for_decision(decision) for decision in selected]
    discovered_rows = [
        {
            "run_id": decision["run_id"],
            "policy_run_id": decision["policy_run_id"],
            "platform_content_id": decision["decision_target_id"],
            "content_discovery_id": f"discovery_{decision['decision_target_id']}",
        }
        for decision in selected
    ]

    assert seen_actions == wanted_actions
    assert _join_feedback(feedback_rows, discovered_rows) == feedback_rows


def test_fake_feedback_join_uses_platform_content_id_not_content_discovery_id():
    decision = _load_historical_decisions("v4_douyin_57663")[0]
    feedback = [_fake_feedback_for_decision(decision)]
    discovered = [
        {
            "run_id": decision["run_id"],
            "policy_run_id": decision["policy_run_id"],
            "platform_content_id": decision["decision_target_id"],
            "content_discovery_id": f"discovery_{decision['decision_target_id']}",
        }
    ]

    assert _join_feedback(feedback, discovered, use_content_discovery_id=True) == []