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

from content_agent.business_modules.walk_engine import _terminal_stage
from content_agent.constants import RUNTIME_RECORD_SCHEMA_VERSION
from content_agent.integrations.walk_strategy_json import WalkStrategyStore


def test_terminal_stage_maps_p5_decisions_to_p6_budget_and_stop_actions():
    result = _terminal_stage(
        {"pattern_execution_id": 581},
        [],
        [
            {
                "platform_content_id": "content_success",
                "search_query_id": "q_001",
                "discovery_start_source": "pattern_itemset",
                "previous_discovery_step": "search_query_direct",
            },
            {
                "platform_content_id": "content_pending",
                "search_query_id": "q_001",
                "discovery_start_source": "pattern_itemset",
                "previous_discovery_step": "search_query_direct",
            },
            {
                "platform_content_id": "content_blocked",
                "search_query_id": "q_001",
                "discovery_start_source": "pattern_itemset",
                "previous_discovery_step": "search_query_direct",
            },
        ],
        [
            _decision("d_001", "content_success", "ADD_TO_CONTENT_POOL", "success"),
            _decision("d_002", "content_pending", "KEEP_CONTENT_FOR_REVIEW", "pending"),
            _decision("d_003", "content_blocked", "REJECT_CONTENT", "rule_blocked"),
        ],
        WalkStrategyStore().load_walk_strategy(),
        "2026-06-11T00:00:00+00:00",
    )

    by_target = {row["decision_target_id"]: row for row in result["walk_actions"]}
    assert by_target["content_success"]["walk_action"] == "commit_asset"
    assert by_target["content_pending"]["walk_action"] == "downgrade_budget"
    assert by_target["content_pending"]["budget_tier"] == "low_budget"
    assert by_target["content_blocked"]["walk_action"] == "stop_path"
    assert by_target["content_blocked"]["walk_status"] == "skipped"


def _decision(decision_id, target_id, action, effect_status):
    return {
        "record_schema_version": RUNTIME_RECORD_SCHEMA_VERSION,
        "run_id": "run_001",
        "policy_run_id": "policy_run_001",
        "decision_id": decision_id,
        "rule_pack_id": "douyin_content_discovery_rule_pack_v1",
        "rule_pack_version": "1.0.0",
        "decision_target_type": "content",
        "decision_target_id": target_id,
        "decision_action": action,
        "decision_reason_code": effect_status,
        "search_query_effect_status": effect_status,
        "decision_input_snapshot_id": f"evidence:{target_id}",
    }


def test_keep_review_author_edge_uses_low_budget(tmp_path):
    from content_agent.business_modules.walk_engine import run_bounded_walk
    from tests.p6_walk_helpers import FakeWalkPlatformClient, build_initial_walk_context

    context = build_initial_walk_context(tmp_path)
    for decision in context["rule_decisions"]:
        decision["decision_action"] = "KEEP_CONTENT_FOR_REVIEW"
        decision["search_query_effect_status"] = "pending"
    client = FakeWalkPlatformClient()

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

    author_actions = [
        row for row in result["walk_actions"] if row["edge_id"] == "author_to_works"
    ]
    assert author_actions
    assert all(row["budget_tier"] == "low_budget" for row in author_actions)


def _walk_configs():
    from content_agent.integrations.walk_graph_json import WalkGraphStore

    store = WalkGraphStore()
    return store.load_policy(), store.load_profile("douyin"), WalkStrategyStore().load_walk_strategy()


def _item(content_id, author_id, *, search_query_id="q_001", has_more=False, cursor=None):
    return {
        "platform_content_id": content_id,
        "platform_author_id": author_id,
        "search_query_id": search_query_id,
        "discovery_start_source": "pattern_itemset",
        "previous_discovery_step": "search_query_direct",
        "has_more": has_more,
        "next_cursor": cursor,
        "tags": [],
    }


class _EmptyAuthorWorksClient:
    def fetch_author_works(self, query):
        return []


def test_author_budget_overflow_emits_budget_exhausted_skip():
    # R8(2026-06-12 拍板): 预算外的合格作者不再静默消失,必须留 budget_exhausted skip。
    # M8D:_expand_authors 改单层 frontier 签名,预算/总闸/去重读 _FrontierContext。
    from content_agent.business_modules.progressive_screening import _ProgressiveContext
    from content_agent.business_modules.walk_engine import _FrontierContext, _expand_authors

    policy, profile, walk_strategy = _walk_configs()
    budget = policy["edge_budgets_by_id"]["author_to_works"]["max_total_actions"]
    total = budget + 2
    items = [_item(f"content_{i}", f"author_{i:03d}") for i in range(1, total + 1)]
    decisions = [
        _decision(f"d_{i:03d}", f"content_{i}", "ADD_TO_CONTENT_POOL", "success")
        for i in range(1, total + 1)
    ]
    ctx = _FrontierContext(
        max_depth=int(policy["global"]["max_depth"]),
        max_total_actions=int(policy["global"]["max_total_actions_per_run"]),
        tag_budget=policy["edge_budgets_by_id"]["hashtag_to_query"]["max_total_actions"],
        author_budget=budget,
        seen_content_ids=set(),
    )
    ctx.depth = 1
    walk_ctx = _ProgressiveContext(
        run_id="run_001",
        policy_run_id="policy_run_001",
        source_context={},
        policy_bundle={},
        platform_client=_EmptyAuthorWorksClient(),
        runtime=None,
        gemini_video_client=None,
        limiter=None,
        archive_dispatcher=None,
        external_seen_content_ids=ctx.seen_content_ids,
    )

    _, actions, _ = _expand_authors(
        list(zip(items, decisions)),
        ctx=ctx,
        walk_ctx=walk_ctx,
        run_id="run_001",
        policy_run_id="policy_run_001",
        policy=policy,
        profile=profile,
        walk_strategy=walk_strategy,
        content_pack={"rule_pack_id": "douyin_content_discovery_rule_pack_v1", "rule_pack_version": "1.0.0"},
        platform_client=_EmptyAuthorWorksClient(),
        limiter=None,
        created_at="2026-06-12T00:00:00+00:00",
    )

    fetched = [row for row in actions if row["walk_status"] == "success"]
    skipped = [row for row in actions if row["reason_code"] == "budget_exhausted"]
    assert len(fetched) == budget
    assert len(skipped) == total - budget
    assert all(row["walk_status"] == "skipped" and row["budget_tier"] == "blocked" for row in skipped)
    assert len({row["walk_action_id"] for row in actions}) == len(actions)


# M8:query_next_page 边已删,翻页归 progressive_screening;原 test_page_budget_overflow 随之移除。