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

"""V3-M4: profile 驱动的平台退化 + walk_actions 指纹基线。

视频号 author_to_works 在 platform_profiles 标 blocked → loop 产显式 skip、不调平台、
不抛错,游走退化为"搜索→内容→tag 回灌"。real_id45 指纹快照由实跑钉死,
是 M5 并发改造"结果与串行一致"的对照基线。
"""

from __future__ import annotations

import json
from pathlib import Path

from tests.gemini_helpers import FakeGeminiVideoClient
from tests.replay_harness import replay_case

_FINGERPRINT_PATH = Path("tests/fixtures/snapshots/real_id45/walk_actions_fingerprint.json")


def _fingerprint(walk_actions):
    # wa_id 含随机 run_id 不能跨次钉死;其确定性由 sha1(run:policy:edge:target:suffix)算法保证。
    return sorted(
        [row["edge_id"], row["from_node_id"], row["to_node_id"], row["walk_action"], row["walk_status"], row["budget_tier"], row.get("reason_code") or ""]
        for row in walk_actions
    )


def test_replay_id45_walk_actions_fingerprint_is_stable(tmp_path):
    # 同输入必须产出同 walk_actions 集合(M5 并发一致性的对照基线;快照由实跑钉死)。
    artifacts = replay_case("real_id45", runtime_root=tmp_path / "rt")
    walk_actions = artifacts.files["walk_actions.jsonl"]
    expected = json.loads(_FINGERPRINT_PATH.read_text(encoding="utf-8"))
    assert _fingerprint(walk_actions) == expected
    assert len({row["walk_action_id"] for row in walk_actions}) == len(walk_actions)


def test_shipinhao_author_edge_blocked_emits_explicit_skip(tmp_path):
    artifacts = replay_case(
        "sph_caihong",
        runtime_root=tmp_path / "rt",
        gemini_video_client=FakeGeminiVideoClient(),
    )
    assert artifacts.state["status"] == "success"
    walk_actions = artifacts.files["walk_actions.jsonl"]

    author_actions = [row for row in walk_actions if row["edge_id"] == "author_to_works"]
    assert author_actions
    assert all(row["walk_status"] == "skipped" for row in author_actions)
    assert all(row["reason_code"] == "edge_blocked_by_platform_profile" for row in author_actions)
    # 退化不是漏抓:无作者作品内容混入。
    assert not [
        item for item in artifacts.files["discovered_content_items.jsonl"]
        if item.get("previous_discovery_step") == "author_works"
    ]