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

"""M8E 核心:多级 frontier 深游走——标签↔作者交替深挖,三道闸(去重/预算/depth)+ 总闸收口。

真实 run 当前全是 depth=1、无多跳链,深度逻辑只能用 mock 合成多层假数据钉死:
depth≤5 / 标签≤5 / 作者≤5 / 总跳≤10 / 三 seen 集去重 / wa_id 唯一 / validate_run pass。
"""

from typing import Any

from content_agent.business_modules import learning_review, result_source_lookup, run_record
from content_agent.business_modules.progressive_screening import _ProgressiveContext
from content_agent.business_modules.run_record.validation import validate_run
from content_agent.business_modules.walk_engine import (
    _FrontierContext,
    _expand_hashtags,
    run_bounded_walk,
)
from content_agent.integrations.walk_graph_json import WalkGraphStore
from content_agent.integrations.walk_strategy_json import WalkStrategyStore
from tests.gemini_helpers import FakeGeminiVideoClient
from tests.p6_walk_helpers import build_initial_walk_context, set_v4_allow_walk

EXPANSION_EDGES = {"hashtag_to_query", "author_to_works"}


def _content(i: int, query: dict[str, Any], *, tags: list[str]) -> dict[str, Any]:
    cid = f"79{i:014d}"
    return {
        "content_discovery_id": f"cd_{i}",
        "search_query_id": query.get("search_query_id", "q"),
        "platform": "douyin",
        "platform_content_id": cid,
        "platform_content_format": "video",
        "description": "deep",
        "platform_author_id": f"auG{i:04d}",
        "author_display_name": "n",
        "statistics": {
            # M11:平台分改 ratio 后,光大点赞不够;给高共鸣占比让平台分过 65 门槛(本意:能继续游走)。
            "digg_count": 100_000,
            "comment_count": 8_000,
            "share_count": 30_000,
            "collect_count": 50_000,
        },
        "tags": tags,
        "score": 72,
        "risk_level": "low",
        "availability": "available",
        "discovery_relation": "fake",
        "discovery_start_source": "pattern_itemset",
        "previous_discovery_step": query.get("previous_discovery_step", "search_query_direct"),
        "content_metadata_source": "fake",
        "platform_raw_payload": {"content_id": cid},
    }


class _BranchingClient:
    """标签搜索与作者作品都带回带新标签+新作者的内容 → 双边分支,预算先于 depth 收口。"""

    def __init__(self) -> None:
        self.search_calls: list[dict[str, Any]] = []
        self.author_calls: list[dict[str, Any]] = []
        self.n = 0

    def _gen(self, query: dict[str, Any]) -> dict[str, Any]:
        self.n += 1
        return _content(self.n, query, tags=[f"tagG{self.n:04d}"])

    def search(self, query: dict[str, Any]) -> list[dict[str, Any]]:
        self.search_calls.append(dict(query))
        if query.get("search_query_generation_method") != "tag_query":
            return []
        return [self._gen(query)]

    def fetch_author_works(self, query: dict[str, Any]) -> list[dict[str, Any]]:
        self.author_calls.append(dict(query))
        return [self._gen(query)]


class _TagChainClient(_BranchingClient):
    """只有标签边带回新内容(每层 1 条),作者边返回空 → 窄链,depth 能逼到天花板 5。"""

    def fetch_author_works(self, query: dict[str, Any]) -> list[dict[str, Any]]:
        self.author_calls.append(dict(query))
        return []


def _seed(tmp_path):
    context = build_initial_walk_context(tmp_path)
    set_v4_allow_walk(context["rule_decisions"][0], True)
    return context


def _set_initial_platform(context: dict[str, Any], platform: str) -> None:
    for item in context["discovered_content_items"]:
        item["platform"] = platform


def _expansion_successes(walk_actions):
    return [
        a for a in walk_actions
        if a["edge_id"] in EXPANSION_EDGES and a["walk_status"] == "success"
    ]


def test_deep_frontier_alternates_edges_within_caps(tmp_path, monkeypatch):
    # 本测试钉死小预算(10/5/5)验"撞顶留痕"机制,与生产 walk_policy 数值解耦(生产改大不影响本测试)。
    from content_agent.integrations import walk_graph_json as _wgj
    _orig_load = _wgj.WalkGraphStore.load_policy

    def _capped(self):
        import copy
        policy = copy.deepcopy(_orig_load(self))  # 深拷贝,别污染共享缓存的 policy
        policy["global"]["max_total_actions_per_run"] = 10
        policy["edge_budgets_by_id"]["hashtag_to_query"]["max_total_actions"] = 5
        policy["edge_budgets_by_id"]["author_to_works"]["max_total_actions"] = 5
        return policy

    monkeypatch.setattr(_wgj.WalkGraphStore, "load_policy", _capped)
    context = _seed(tmp_path)
    client = _BranchingClient()

    result = run_bounded_walk(platform_client=client, **context)
    actions = result["walk_actions"]
    successes = _expansion_successes(actions)

    # 标签↔作者都参与,且跨多层(depth 1/2/3 均有向外游走)。
    edges = {a["edge_id"] for a in successes}
    assert edges == EXPANSION_EDGES
    depths = {a["depth"] for a in successes}
    assert {1, 2, 3} <= depths

    # 三道闸 + 总闸(本测试钉死的小预算):depth≤5、标签≤5、作者≤5、总向外跳≤10。
    assert max(depths) <= 5
    tag_jumps = [a for a in successes if a["edge_id"] == "hashtag_to_query"]
    author_jumps = [a for a in successes if a["edge_id"] == "author_to_works"]
    assert len(tag_jumps) <= 5
    assert len(author_jumps) <= 5
    assert len(successes) <= 10

    # 撞顶留痕:预算耗尽有 budget_exhausted skip。
    assert any(a["reason_code"] == "budget_exhausted" for a in actions)

    # id 唯一:wa_id / decision_id 全 run 不撞(深游走历史事故面)。
    wa_ids = [a["walk_action_id"] for a in actions]
    assert len(wa_ids) == len(set(wa_ids))
    decision_ids = [d["decision_id"] for d in result["rule_decisions"]]
    assert len(decision_ids) == len(set(decision_ids))


def test_deep_frontier_reaches_max_depth_then_stops(tmp_path):
    context = _seed(tmp_path)
    client = _TagChainClient()

    result = run_bounded_walk(platform_client=client, **context)
    successes = _expansion_successes(result["walk_actions"])
    depths = {a["depth"] for a in successes}

    # 窄链每层 1 条 → 标签链逐层加深到 max_depth=5;depth 不得越界,挂视频继承父+1。
    assert max(depths) == 5
    assert all(d <= 5 for d in depths)


def test_kuaishou_supported_author_edge_fetches_author_works(tmp_path):
    context = _seed(tmp_path)
    _set_initial_platform(context, "kuaishou")
    client = _BranchingClient()

    result = run_bounded_walk(platform_client=client, **context)

    assert client.author_calls
    assert any(
        action["edge_id"] == "author_to_works" and action["walk_status"] == "success"
        for action in result["walk_actions"]
    )
    assert any(
        item.get("previous_discovery_step") == "author_works"
        for item in result["discovered_content_items"]
    )


def test_shipinhao_blocked_author_edge_does_not_call_client(tmp_path):
    context = _seed(tmp_path)
    _set_initial_platform(context, "shipinhao")
    client = _BranchingClient()

    result = run_bounded_walk(platform_client=client, **context)

    assert client.author_calls == []
    assert any(
        action["edge_id"] == "author_to_works"
        and action["walk_status"] == "skipped"
        and action["reason_code"] == "edge_blocked_by_platform_profile"
        for action in result["walk_actions"]
    )


def test_deep_frontier_dedups_repeated_content_author_tag(tmp_path):
    # 同内容/作者/标签反复出现:三 seen 集去重,不成环、不爆炸。
    context = _seed(tmp_path)

    class _RepeatClient(_BranchingClient):
        def _gen(self, query):
            # 固定返回同一条内容(同 content_id / 作者 / 标签)。
            return _content(1, query, tags=["tagG0001"])

    client = _RepeatClient()
    result = run_bounded_walk(platform_client=client, **context)

    content_ids = [i["platform_content_id"] for i in result["discovered_content_items"]]
    assert len(content_ids) == len(set(content_ids))  # 无重复内容
    successes = _expansion_successes(result["walk_actions"])
    assert len(successes) <= 30


def test_global_action_cap_emits_skip(tmp_path):
    # 总闸耗尽:即便边预算仍有余,向外跳也被 global_action_cap_reached 拦截。
    context = _seed(tmp_path)
    item = context["discovered_content_items"][0]
    decision = context["rule_decisions"][0]
    ctx = _FrontierContext(
        max_depth=5, max_total_actions=2, tag_budget=5, author_budget=5, seen_content_ids=set()
    )
    ctx.depth = 1
    ctx.global_walk_action_count = 2  # 已达总闸
    walk_ctx = _ProgressiveContext(
        run_id=context["run_id"], policy_run_id=context["policy_run_id"],
        source_context=context["source_context"], policy_bundle=context["policy_bundle"],
        platform_client=_BranchingClient(), runtime=context["runtime"],
        gemini_video_client=FakeGeminiVideoClient(), limiter=None, archive_dispatcher=None,
        external_seen_content_ids=ctx.seen_content_ids,
    )
    store = WalkGraphStore()
    brought, actions, queries = _expand_hashtags(
        [(item, decision)], ctx=ctx, walk_ctx=walk_ctx,
        run_id=context["run_id"], policy_run_id=context["policy_run_id"],
        policy=store.load_policy(), profile=store.load_profile("douyin"),
        walk_strategy=WalkStrategyStore().load_walk_strategy(),
        content_pack={"rule_pack_id": "douyin_content_discovery_rule_pack_v1", "rule_pack_version": "1.0.0"},
        created_at="2026-06-18T00:00:00+00:00",
    )
    assert brought == []
    assert queries == []
    assert [a["reason_code"] for a in actions] == ["global_action_cap_reached"]


def test_deep_frontier_validate_run_passes(tmp_path):
    context = _seed(tmp_path)
    client = _BranchingClient()

    result = run_bounded_walk(platform_client=client, **context)

    record = run_record.run(
        context["run_id"], context["policy_run_id"], result["search_queries"],
        result["discovered_content_items"], result["rule_decisions"],
        result["source_path_record_basis"], context["policy_bundle"], context["runtime"],
        walk_actions=result["walk_actions"],
    )
    result_source_lookup.run(
        context["run_id"], context["policy_run_id"], context["policy_bundle"],
        result["discovered_content_items"], result["content_media_records"],
        result["rule_decisions"], record["source_path_records"], record["search_clues"],
        context["runtime"],
    )
    learning_review.run(context["run_id"], context["policy_run_id"], context["runtime"])
    validation = validate_run(context["run_id"], context["runtime"])
    fails = [f for f in validation["findings"] if f["level"] == "fail"]
    assert validation["status"] == "pass", fails