content-find-agent-new/content_agent/business_modules/walk_engine.py

"""有界游走(V3-M4):配置驱动的统一 frontier 流程,取代三段硬编码。

每条边走同一套闸:platform_profiles 是否 supported(blocked 显式 skip 不调用平台)
→ walk_policy.edge_permissions 按判定结果放行(取代 _can_expand_from_decision 硬编码)
→ walk_policy.edge_budgets 预算(取代 [:2]/[:3]/>=1 散落硬限;预算耗尽静默,对齐 v1 行为)。
终端边(commit/downgrade/stop)与 3 类血缘并入本模块终端阶段(原 plan_walk 节点已删)。
"""

from __future__ import annotations

import hashlib
import time
from datetime import datetime, timezone
from typing import Any, Callable

from content_agent.business_modules import walk_strategy as walk_terminal
from content_agent.business_modules.progressive_screening import (
    SearchCallLimiter,
    _ProgressiveContext,
    make_search_limiter,
)
from content_agent.constants import RUNTIME_RECORD_SCHEMA_VERSION
from content_agent.integrations.walk_graph_json import (
    WalkGraphStore,
    edge_budgets_for_platform,
    edge_permission,
    edge_supported,
)
from content_agent.integrations.walk_strategy_json import WalkStrategyStore
from content_agent.interfaces import (
    GeminiVideoClient,
    PlatformSearchClient,
    RuntimeFileStore,
)
from content_agent.record_payload import with_raw_payload


V4_SCORECARD_SCHEMA_VERSION = "v4_scorecard.v1"
V4_ALLOW_WALK_DENIED_REASON = "v4_allow_walk_denied"
V4_WALK_GATE_EDGES = {"hashtag_to_query", "author_to_works"}

BLOCKED_TAG_TERMS = {
    "h" + "ot",
    "trend" + "ing",
    "热" + "点",
    "热" + "榜",
    "养" + "号",
    "无" + "限",
}


def run_bounded_walk(
    *,
    run_id: str,
    policy_run_id: str,
    pattern_seed_pack: dict[str, Any],
    source_context: dict[str, Any],
    search_queries: list[dict[str, Any]],
    discovered_content_items: list[dict[str, Any]],
    content_media_records: list[dict[str, Any]],
    evidence_bundles: list[dict[str, Any]],
    rule_decisions: list[dict[str, Any]],
    policy_bundle: dict[str, Any],
    platform_client: PlatformSearchClient,
    runtime: RuntimeFileStore,
    gemini_video_client: GeminiVideoClient,
    query_gate: Callable[[str], bool] | None = None,
    archive_dispatcher: Any | None = None,
) -> dict[str, list[dict[str, Any]]]:
    created_at = datetime.now(timezone.utc).isoformat()
    walk_strategy = WalkStrategyStore().load_walk_strategy()
    store = WalkGraphStore()
    policy = store.load_policy()
    max_depth = int(policy["global"]["max_depth"])
    platform = next(
        (item["platform"] for item in discovered_content_items if item.get("platform")),
        "douyin",
    )
    max_total_actions = int(policy["global"]["max_total_actions_per_run"])
    budgets = edge_budgets_for_platform(policy, platform)
    policy = {**policy, "edge_budgets_by_id": budgets}
    profile = store.load_profile(platform)
    content_pack = {
        "rule_pack_id": policy_bundle["rule_pack_id"],
        "rule_pack_version": policy_bundle["rule_pack_version"],
    }
    context = {
        "search_queries": list(search_queries),
        "discovered_content_items": list(discovered_content_items),
        "content_media_records": list(content_media_records),
        "evidence_bundles": list(evidence_bundles),
        "rule_decisions": list(rule_decisions),
        "walk_actions": [],
    }

    # 全 run content 去重集(首轮 + 跨层),与 walk_ctx 共享同一对象。
    seen_content_ids = {
        str(item["platform_content_id"])
        for item in discovered_content_items
        if item.get("platform_content_id")
    }
    # 游走限速器:real client 才挂;抖音 10-12s 随机,其他平台默认 30s。
    # 标签搜索经 walk_ctx._fetch_page 限速,作者作品在 _expand_authors 调 fetch 前限速,共享同一实例。
    limiter = (
        make_search_limiter(platform)
        if getattr(platform_client, "requires_progressive_search_rate_limit", False)
        else None
    )
    # M8C 统一搜索单元:游走自有 _ProgressiveContext(独立累积 + 注入首轮 seen + index 基线)。
    walk_ctx = _ProgressiveContext(
        run_id=run_id,
        policy_run_id=policy_run_id,
        source_context=source_context,
        policy_bundle=policy_bundle,
        platform_client=platform_client,
        runtime=runtime,
        gemini_video_client=gemini_video_client,
        limiter=limiter,
        portrait_limiter=None,
        archive_dispatcher=archive_dispatcher,
        external_seen_content_ids=seen_content_ids,
        recall_index_base=len(discovered_content_items),
        decision_index_base=len(rule_decisions),
        platform=platform,
    )
    fctx = _FrontierContext(
        max_depth=max_depth,
        max_total_actions=max_total_actions,
        tag_budget=budgets["hashtag_to_query"]["max_total_actions"],
        author_budget=budgets["author_to_works"]["max_total_actions"],
        seen_content_ids=seen_content_ids,
    )

    # 初始 frontier = 首轮全部(item, decision):允许的向外游走,被拒的留 skip 留痕。
    decision_by_content_id = _decision_by_content_id(context["rule_decisions"])
    frontier = [
        (item, decision_by_content_id[item["platform_content_id"]])
        for item in discovered_content_items
        if item.get("platform_content_id") in decision_by_content_id
    ]

    # M8D 多级 frontier 深游走:标签↔作者任意交替,depth/总闸/各边预算/三 seen 集收口。
    while frontier and fctx.depth < max_depth and fctx.global_left():
        fctx.depth += 1
        tag_back, tag_actions, tag_queries = _expand_hashtags(
            frontier, ctx=fctx, walk_ctx=walk_ctx, run_id=run_id, policy_run_id=policy_run_id,
            policy=policy, profile=profile, walk_strategy=walk_strategy,
            content_pack=content_pack, created_at=created_at, query_gate=query_gate,
        )
        author_back, author_actions, author_queries = _expand_authors(
            frontier, ctx=fctx, walk_ctx=walk_ctx, run_id=run_id, policy_run_id=policy_run_id,
            policy=policy, profile=profile, walk_strategy=walk_strategy,
            content_pack=content_pack, platform_client=platform_client,
            limiter=limiter, created_at=created_at,
        )
        context["walk_actions"].extend(tag_actions)
        context["walk_actions"].extend(author_actions)
        new_queries = tag_queries + author_queries
        if new_queries:
            runtime.append_jsonl(run_id, "search_queries.jsonl", new_queries)
            context["search_queries"].extend(new_queries)
        frontier = tag_back + author_back

    # 游走带回内容统一一次落盘(append 到首轮已写文件),并并入 context 供终端/记录。
    # write_runtime 会 drain 已完成的 OSS 归档 + enable 后续异步写;之后 close(shutdown wait=False),
    # 与首轮 progressive_screening.run 的生命周期一致——否则游走带回视频拿不到 OSS、Gemini 无源→全技术重试。
    walk_ctx.write_runtime()
    walk_ctx.close_archive_dispatcher()
    context["discovered_content_items"].extend(walk_ctx.discovered_content_items)
    context["content_media_records"].extend(walk_ctx.content_media_records)
    context["evidence_bundles"].extend(walk_ctx.evidence_bundles)
    context["rule_decisions"].extend(walk_ctx.rule_decisions)

    terminal = _terminal_stage(
        pattern_seed_pack,
        context["search_queries"],
        context["discovered_content_items"],
        context["rule_decisions"],
        walk_strategy,
        created_at,
    )
    context["walk_actions"].extend(terminal["walk_actions"])
    context["source_path_record_basis"] = terminal["source_path_record_basis"]
    return context


class _FrontierContext:
    """游走 frontier 的全 run 级状态机(进程内,不落库)。

    三个全 run 去重集 + 各边运行预算 + 总闸 + depth 贯穿所有层,任一闸单独都能收口、不成环。
    content_id 去重集与 walk 级 _ProgressiveContext 的 external_seen_content_ids 共享同一对象(单一真相);
    recall/decision id 与 index 由 walk_ctx 累积自动推进(消除手动推进事故源)。
    """

    def __init__(
        self,
        *,
        max_depth: int,
        max_total_actions: int,
        tag_budget: int,
        author_budget: int,
        seen_content_ids: set[str],
    ) -> None:
        self.max_depth = max_depth
        self.max_total_actions = max_total_actions
        self.remaining_budget = {
            "hashtag_to_query": tag_budget,
            "author_to_works": author_budget,
        }
        self.seen_content_ids = seen_content_ids
        self.seen_query_keys: set[str] = set()
        self.seen_author_ids: set[str] = set()
        self.global_walk_action_count = 0
        self.depth = 0

    def budget_left(self, edge_id: str) -> bool:
        return self.remaining_budget.get(edge_id, 0) > 0

    def global_left(self) -> bool:
        return self.global_walk_action_count < self.max_total_actions

    def consume(self, edge_id: str) -> None:
        # 成功向外游走一次(标签跳或作者跳):扣边预算 + 累计总闸。失败的跳不调用本方法、不扣预算。
        self.remaining_budget[edge_id] -= 1
        self.global_walk_action_count += 1

    def budget_snapshot(self) -> dict[str, int]:
        return dict(self.remaining_budget)


def _frontier_pairs(batch: dict[str, list[dict[str, Any]]]) -> list[tuple[dict[str, Any], dict[str, Any]]]:
    """把统一搜索单元带回的新内容配成 (item, decision),供下一层 frontier。"""
    decision_by_id = {d["decision_target_id"]: d for d in batch["rule_decisions"]}
    pairs: list[tuple[dict[str, Any], dict[str, Any]]] = []
    for item in batch["discovered_content_items"]:
        decision = decision_by_id.get(item.get("platform_content_id"))
        if decision is not None:
            pairs.append((item, decision))
    return pairs


def _expand_hashtags(
    frontier: list[tuple[dict[str, Any], dict[str, Any]]],
    *,
    ctx: _FrontierContext,
    walk_ctx: _ProgressiveContext,
    run_id: str,
    policy_run_id: str,
    policy: dict[str, Any],
    profile: dict[str, Any],
    walk_strategy: dict[str, Any],
    content_pack: dict[str, Any],
    created_at: str,
    query_gate: Callable[[str], bool] | None = None,
) -> tuple[
    list[tuple[dict[str, Any], dict[str, Any]]],
    list[dict[str, Any]],
    list[dict[str, Any]],
]:
    """单层标签扩展:对本层每个 allow_walk 视频的标签建 query,走统一搜索单元(前3→翻页≤3页)。

    回灌链 video_to_hashtag → hashtag_to_query:被拒留 skip;预算/总闸/去重读 ctx 跨层运行扣减。
    返回 (带回的 (item, decision) 列表, walk_actions, 新建 search_query 行)。
    """
    walk_actions: list[dict[str, Any]] = []
    brought_back: list[tuple[dict[str, Any], dict[str, Any]]] = []
    new_queries: list[dict[str, Any]] = []
    if not (edge_supported(profile, "video_to_hashtag") and edge_supported(profile, "hashtag_to_query")):
        return brought_back, walk_actions, new_queries
    tag_binding, _ = _resolve_edge_binding("hashtag_to_query", walk_strategy)
    for item, decision in frontier:
        if _edge_permission_for(decision, "video_to_hashtag", policy) == "deny":
            if item.get("tags"):
                reason_code = (
                    V4_ALLOW_WALK_DENIED_REASON
                    if _v4_walk_gate_denied(decision)
                    else
                    "review_tag_expansion_disabled"
                    if decision.get("decision_action") == "KEEP_CONTENT_FOR_REVIEW"
                    else "blocked_by_rule_decision"
                )
                walk_actions.append(
                    _tag_skip_action(
                        run_id, policy_run_id, item, decision, created_at,
                        reason_code=reason_code, suffix="tags", depth=ctx.depth,
                        tag_binding=tag_binding, content_pack=content_pack,
                        raw_extra=_decision_context(
                            decision, denied=reason_code == V4_ALLOW_WALK_DENIED_REASON
                        ),
                    )
                )
            continue
        for tag in item.get("tags") or []:
            normalized = str(tag).lstrip("#").strip()
            if not normalized or normalized in ctx.seen_query_keys or _blocked_tag(normalized):
                continue
            # M9D Gate 2:非抖音标签 query 复用 50+ 判定(query_gate 仅非抖音注入);判否不搜。
            if query_gate is not None and not query_gate(normalized):
                continue
            if not ctx.budget_left("hashtag_to_query"):
                # R8:合格标签排不上队,补一条 budget_exhausted skip(每内容至多一条)。
                walk_actions.append(
                    _tag_skip_action(
                        run_id, policy_run_id, item, decision, created_at,
                        reason_code="budget_exhausted", suffix="tag_budget", depth=ctx.depth,
                        tag_binding=tag_binding, content_pack=content_pack,
                        raw_extra={**_decision_context(decision), "hashtag": normalized},
                    )
                )
                break
            if not ctx.global_left():
                walk_actions.append(
                    _tag_skip_action(
                        run_id, policy_run_id, item, decision, created_at,
                        reason_code="global_action_cap_reached", suffix="tag_global", depth=ctx.depth,
                        tag_binding=tag_binding, content_pack=content_pack,
                        raw_extra={**_decision_context(decision), "hashtag": normalized},
                    )
                )
                break
            ctx.seen_query_keys.add(normalized)
            tag_query = _search_query_row(
                run_id, policy_run_id, f"tag_{_short_hash(normalized)}", normalized, "tag_query",
                item.get("discovery_start_source", "pattern_itemset"), "hashtag_to_query", created_at,
                raw_extra={
                    "hashtag": normalized,
                    "source_content_id": item.get("platform_content_id"),
                    **_decision_context(decision),
                },
            )
            new_queries.append(tag_query)
            fail_before = len(walk_ctx.query_failures)
            batch = walk_ctx.process_query(tag_query)
            failed = len(walk_ctx.query_failures) > fail_before
            if not failed:
                ctx.consume("hashtag_to_query")  # 失败的标签游走不扣预算
            walk_actions.append(
                _tag_query_action(
                    tag_query, created_at, status="failed" if failed else "success",
                    depth=ctx.depth, ctx=ctx, walk_strategy=walk_strategy, content_pack=content_pack,
                )
            )
            if not failed:
                brought_back.extend(_frontier_pairs(batch))
    return brought_back, walk_actions, new_queries


def _tag_skip_action(
    run_id: str,
    policy_run_id: str,
    item: dict[str, Any],
    decision: dict[str, Any] | None,
    created_at: str,
    *,
    reason_code: str,
    suffix: str,
    depth: int,
    tag_binding: dict[str, Any],
    content_pack: dict[str, Any],
    raw_extra: dict[str, Any],
) -> dict[str, Any]:
    return _walk_action(
        run_id, policy_run_id,
        _walk_action_id(run_id, policy_run_id, "hashtag_to_query", item["platform_content_id"], suffix, depth=depth),
        "hashtag_to_query", "tag_query", "Content", item["platform_content_id"],
        "SearchQuery", "tag_query_skipped", "create_tag_query", "skipped", created_at,
        depth=depth, reason_code=reason_code, budget_tier="blocked",
        rule_pack_binding=tag_binding,
        rule_pack_execution=_execution_record(decision, content_pack_id=content_pack["rule_pack_id"]),
        fallback_rule_pack=content_pack, raw_extra=raw_extra,
    )


def _tag_query_action(
    tag_query: dict[str, Any],
    created_at: str,
    *,
    status: str,
    depth: int,
    ctx: _FrontierContext,
    walk_strategy: dict[str, Any],
    content_pack: dict[str, Any],
) -> dict[str, Any]:
    binding, _ = _resolve_edge_binding("hashtag_to_query", walk_strategy)
    raw_extra = dict(_walk_gate_context_from_query_row(tag_query))
    raw_extra.update(_walk_runtime_extra(ctx))
    return _walk_action(
        tag_query["run_id"], tag_query["policy_run_id"],
        _walk_action_id(
            tag_query["run_id"], tag_query["policy_run_id"], "hashtag_to_query",
            tag_query["search_query_id"], "query", depth=depth,
        ),
        "hashtag_to_query", "query", "SearchQuery",
        tag_query.get("raw_payload", {}).get("hashtag") or tag_query["search_query_id"],
        "SearchQuery", tag_query["search_query_id"], "create_tag_query", status, created_at,
        depth=depth,
        rule_pack_binding=binding,
        rule_pack_execution={
            "executed": True,
            "executed_rule_pack_id": content_pack["rule_pack_id"],
            "reason": "content_decision_reused_for_walk_gate",
        },
        fallback_rule_pack=content_pack, raw_extra=raw_extra,
    )


def _expand_authors(
    frontier: list[tuple[dict[str, Any], dict[str, Any]]],
    *,
    ctx: _FrontierContext,
    walk_ctx: _ProgressiveContext,
    run_id: str,
    policy_run_id: str,
    policy: dict[str, Any],
    profile: dict[str, Any],
    walk_strategy: dict[str, Any],
    content_pack: dict[str, Any],
    platform_client: PlatformSearchClient,
    limiter: SearchCallLimiter | None,
    created_at: str,
) -> tuple[
    list[tuple[dict[str, Any], dict[str, Any]]],
    list[dict[str, Any]],
    list[dict[str, Any]],
]:
    """单层作者扩展:对本层每个 allow_walk 视频的作者拉作品(列表+截断,不分页),复用判定路径。

    本层按作者去重(首次出现代表),跨层经 ctx.seen_author_ids 去重;预算/总闸/content 去重读 ctx。
    返回 (带回的 (item, decision) 列表, walk_actions, 新建 search_query 行)。
    """
    author_budgets = policy["edge_budgets_by_id"]["author_to_works"]
    binding, _ = _resolve_edge_binding("author_to_works", walk_strategy)
    walk_actions: list[dict[str, Any]] = []
    brought_back: list[tuple[dict[str, Any], dict[str, Any]]] = []
    new_queries: list[dict[str, Any]] = []
    unique_frontier = _unique_author_frontier(frontier)

    # 平台不支持作者边(如视频号 blogger blocked):本层每个唯一作者 skip 留痕、不调用平台。
    if not edge_supported(profile, "author_to_works"):
        for item, decision in unique_frontier:
            author_id = item.get("platform_author_id")
            if not author_id:
                continue
            walk_actions.append(
                _author_walk_action(
                    run_id, policy_run_id, author_id, "skipped", created_at,
                    depth=ctx.depth, reason_code="edge_blocked_by_platform_profile",
                    budget_tier="blocked", binding=binding, decision=decision,
                    content_pack=content_pack, raw_extra=_author_raw_extra(decision, ctx),
                )
            )
        return brought_back, walk_actions, new_queries

    fetch_author_works = getattr(platform_client, "fetch_author_works", None) or getattr(
        platform_client, "author_works", None
    )
    if not callable(fetch_author_works):
        return brought_back, walk_actions, new_queries

    for item, decision in unique_frontier:
        author_id = item.get("platform_author_id")
        if not author_id or author_id in ctx.seen_author_ids:
            continue
        permission = _edge_permission_for(decision, "author_to_works", policy)
        if permission == "deny":
            reason_code = (
                V4_ALLOW_WALK_DENIED_REASON
                if _v4_walk_gate_denied(decision)
                else "blocked_by_rule_decision"
            )
            walk_actions.append(
                _author_walk_action(
                    run_id, policy_run_id, author_id, "skipped", created_at,
                    depth=ctx.depth, reason_code=reason_code, budget_tier="blocked",
                    binding=binding, decision=decision, content_pack=content_pack,
                    raw_extra=_author_raw_extra(decision, ctx),
                )
            )
            continue
        if not ctx.budget_left("author_to_works"):
            walk_actions.append(
                _author_walk_action(
                    run_id, policy_run_id, author_id, "skipped", created_at,
                    depth=ctx.depth, reason_code="budget_exhausted", budget_tier="blocked",
                    binding=binding, decision=decision, content_pack=content_pack,
                    raw_extra=_author_raw_extra(decision, ctx),
                )
            )
            continue
        if not ctx.global_left():
            walk_actions.append(
                _author_walk_action(
                    run_id, policy_run_id, author_id, "skipped", created_at,
                    depth=ctx.depth, reason_code="global_action_cap_reached", budget_tier="blocked",
                    binding=binding, decision=decision, content_pack=content_pack,
                    raw_extra=_author_raw_extra(decision, ctx),
                )
            )
            break
        # 走作者跳:本层标记作者已走,扣预算 + 累计总闸。
        ctx.seen_author_ids.add(author_id)
        budget_tier = "low_budget" if permission == "allow_low_budget" else "normal"
        author_query_id = f"author_{_short_hash(author_id)}"
        author_fetch_started_at = time.monotonic()
        waited_seconds = limiter.wait() if limiter is not None else 0.0
        request_started_at = time.monotonic()
        try:
            works = fetch_author_works(
                {
                    "platform_author_id": author_id,
                    "search_query_id": author_query_id,
                    "discovery_start_source": item.get("discovery_start_source", "pattern_itemset"),
                }
            )
            request_ended_at = time.monotonic()
        except Exception as exc:
            # 失败的作者游走不扣预算(作者已 seen,不会重试)
            walk_actions.append(
                _author_walk_action(
                    run_id, policy_run_id, author_id, "failed", created_at,
                    depth=ctx.depth, reason_code=type(exc).__name__, budget_tier=budget_tier,
                    binding=binding, decision=decision, content_pack=content_pack,
                    raw_extra=_author_raw_extra(decision, ctx),
                )
            )
            continue
        ctx.consume("author_to_works")
        walk_actions.append(
            _author_walk_action(
                run_id, policy_run_id, author_id, "success", created_at,
                depth=ctx.depth, budget_tier=budget_tier, binding=binding, decision=decision,
                content_pack=content_pack, raw_extra=_author_raw_extra(decision, ctx),
            )
        )
        # 作者近期作品天然可能含首轮/已发现的同一条视频;先按 ctx.seen_content_ids 去重再截断 max_works_per_author。
        # (不去重会撞 uk_ca_items_run_policy_content 唯一索引,真实 E2E v1_run_e6ba21f7543b 实证。)
        discovery_start_source = item.get("discovery_start_source", "pattern_itemset")
        fresh_works = [
            {
                **work,
                "search_query_id": work.get("search_query_id") or author_query_id,
                "discovery_start_source": work.get("discovery_start_source") or discovery_start_source,
                "previous_discovery_step": "author_works",
            }
            for work in works
            if str(work.get("platform_content_id") or "")
            and str(work.get("platform_content_id")) not in ctx.seen_content_ids
        ][: author_budgets["max_works_per_author"]]
        if not fresh_works:
            continue
        author_query = _search_query_row(
            run_id, policy_run_id, author_query_id, f"author:{author_id}", "author_works",
            discovery_start_source, "author_to_works", created_at,
            raw_extra={"platform_author_id": author_id},
        )
        batch = walk_ctx.process_prefetched_batch(
            author_query,
            fresh_works,
            search_duration_ms=_elapsed_ms(author_fetch_started_at, request_ended_at),
            search_wait_ms=int(waited_seconds * 1000),
            search_request_duration_ms=_elapsed_ms(request_started_at, request_ended_at),
            search_timing_source="author_works_fetch",
        )
        if batch["discovered_content_items"]:
            # 血缘补全:作者作品内容引用的合成 query id 必须真实存在于 search_queries,
            # 否则 validate_run 断链(真实 E2E v1_run_3a3bc9f0d72d 实证:missing_search_query_ref 等 8 条 fail)。
            new_queries.append(author_query)
            brought_back.extend(_frontier_pairs(batch))
    return brought_back, walk_actions, new_queries


def _unique_author_frontier(
    frontier: list[tuple[dict[str, Any], dict[str, Any]]]
) -> list[tuple[dict[str, Any], dict[str, Any]]]:
    """本层按 platform_author_id 去重,保留首次出现的 (item, decision)。"""
    seen: set[str] = set()
    unique: list[tuple[dict[str, Any], dict[str, Any]]] = []
    for item, decision in frontier:
        author_id = item.get("platform_author_id")
        if not author_id or author_id in seen:
            continue
        seen.add(author_id)
        unique.append((item, decision))
    return unique


def _author_raw_extra(decision: dict[str, Any] | None, ctx: _FrontierContext) -> dict[str, Any]:
    return {**_decision_context(decision), **_walk_runtime_extra(ctx)}


def _walk_runtime_extra(ctx: _FrontierContext) -> dict[str, Any]:
    return {
        "walk_depth": ctx.depth,
        "frontier_round": ctx.depth,
        "remaining_budget_snapshot": ctx.budget_snapshot(),
        "global_walk_action_count": ctx.global_walk_action_count,
    }


def _author_walk_action(
    run_id: str,
    policy_run_id: str,
    author_id: str,
    walk_status: str,
    created_at: str,
    *,
    depth: int = 1,
    budget_tier: str,
    binding: dict[str, Any],
    decision: dict[str, Any] | None,
    content_pack: dict[str, Any],
    reason_code: str | None = None,
    raw_extra: dict[str, Any] | None = None,
) -> dict[str, Any]:
    return _walk_action(
        run_id,
        policy_run_id,
        _walk_action_id(run_id, policy_run_id, "author_to_works", author_id, "works", depth=depth),
        "author_to_works",
        "author",
        "Author",
        author_id,
        "AuthorWorksPage",
        author_id,
        "fetch_author_works",
        walk_status,
        created_at,
        depth=depth,
        reason_code=reason_code,
        budget_tier=budget_tier,
        rule_pack_binding=binding,
        rule_pack_execution=_execution_record(decision, content_pack_id=content_pack["rule_pack_id"]),
        fallback_rule_pack=content_pack,
        raw_extra=raw_extra if raw_extra is not None else _decision_context(decision),
    )


def _terminal_stage(
    pattern_seed_pack: dict[str, Any],
    search_queries: list[dict[str, Any]],
    discovered_content_items: list[dict[str, Any]],
    decisions: list[dict[str, Any]],
    walk_strategy: dict[str, Any],
    created_at: str,
) -> dict[str, list[dict[str, Any]]]:
    """终端边(decision_to_asset/budget_downgrade/path_stop)+ 3 类血缘(原 plan_walk 语义)。"""
    binding_by_edge = _binding_by_edge_id(walk_strategy)
    decision_by_target_id = {decision["decision_target_id"]: decision for decision in decisions}
    walk_actions: list[dict[str, Any]] = []
    source_path_record_basis: list[dict[str, Any]] = []

    for search_query in search_queries:
        query_source_ref = search_query.get("pattern_seed_ref") or {}
        source_path_record_basis.append(
            {
                "policy_run_id": search_query["policy_run_id"],
                "record_schema_version": search_query["record_schema_version"],
                "from_node_type": "PatternSeed",
                "from_node_id": pattern_seed_pack["pattern_execution_id"],
                "to_node_type": "SearchQuery",
                "to_node_id": search_query["search_query_id"],
                "source_path_type": "pattern_to_search_query",
                "rule_pack_id": None,
                "decision_id": None,
                "discovery_start_source": search_query["discovery_start_source"],
                "previous_discovery_step": search_query["previous_discovery_step"],
                "origin_path_id": f"pattern_to_search_query:{search_query['search_query_id']}",
                "source_evidence_ref": "source_context.json#ext_data.evidence_pack",
                "query_source_type": query_source_ref.get("query_source_type"),
                "query_source_ref_id": query_source_ref.get("query_source_ref_id"),
                "query_source_text": query_source_ref.get("query_source_text")
                or search_query.get("search_query"),
                "query_source_rank": query_source_ref.get("query_source_rank"),
                "query_source_refs": search_query.get("query_source_refs")
                or search_query.get("raw_payload", {}).get("query_source_refs", []),
            }
        )

    for item in discovered_content_items:
        # 无判定内容不产终端动作;判定覆盖完整性由 validate_run 把关,这里不掩盖。
        decision = decision_by_target_id.get(item["platform_content_id"])
        if not decision:
            continue
        decision_action = walk_terminal._action_for_decision(decision["decision_action"])
        binding = binding_by_edge.get(decision_action["edge_id"]) or {}
        execution = {
            "executed": True,
            "executed_rule_pack_id": decision["rule_pack_id"],
            "reason": "content_decision_reused_for_walk_gate",
        }
        walk_action_id = walk_terminal._walk_action_id(
            decision["run_id"],
            decision["policy_run_id"],
            decision_action["edge_id"],
            item["platform_content_id"],
            decision["decision_id"],
        )
        query_sources = item.get("query_sources") or [
            {
                "search_query_id": item["search_query_id"],
                "search_query": item.get("search_query"),
                "search_query_generation_method": item.get("search_query_generation_method"),
            }
        ]
        for query_source in query_sources:
            search_query_id = query_source["search_query_id"]
            source_path_record_basis.append(
                {
                    "policy_run_id": decision["policy_run_id"],
                    "record_schema_version": decision["record_schema_version"],
                    "from_node_type": "SearchQuery",
                    "from_node_id": search_query_id,
                    "to_node_type": "Content",
                    "to_node_id": item["platform_content_id"],
                    "source_path_type": "search_query_to_content",
                    "rule_pack_id": decision["rule_pack_id"],
                    "decision_id": decision["decision_id"],
                    "discovery_start_source": item["discovery_start_source"],
                    "previous_discovery_step": item["previous_discovery_step"],
                    "origin_path_id": (
                        f"search_query_to_content:{search_query_id}:"
                        f"{item['platform_content_id']}"
                    ),
                    "source_evidence_ref": decision["decision_input_snapshot_id"],
                    "walk_action_id": walk_action_id,
                    "rule_pack_binding": binding,
                    "rule_pack_execution": execution,
                }
            )
        walk_actions.append(
            walk_terminal._walk_action_row(
                decision, item, decision_action, walk_action_id, created_at, binding, execution
            )
        )
        if decision["decision_action"] == "ADD_TO_CONTENT_POOL":
            source_path_record_basis.append(
                {
                    "policy_run_id": decision["policy_run_id"],
                    "record_schema_version": decision["record_schema_version"],
                    "from_node_type": "RuleDecision",
                    "from_node_id": decision["decision_id"],
                    "to_node_type": "ContentAsset",
                    "to_node_id": item["platform_content_id"],
                    "source_path_type": "decision_to_asset",
                    "rule_pack_id": decision["rule_pack_id"],
                    "decision_id": decision["decision_id"],
                    "discovery_start_source": item["discovery_start_source"],
                    "previous_discovery_step": "asset_commit",
                    "origin_path_id": (
                        f"decision_to_asset:{decision['decision_id']}:"
                        f"{item['platform_content_id']}"
                    ),
                    "source_evidence_ref": decision["decision_input_snapshot_id"],
                    "walk_action_id": walk_action_id,
                    "rule_pack_binding": binding,
                    "rule_pack_execution": execution,
                }
            )

    return {"walk_actions": walk_actions, "source_path_record_basis": source_path_record_basis}


def _search_query_row(
    run_id: str,
    policy_run_id: str,
    search_query_id: str,
    search_query: str,
    method: str,
    discovery_start_source: str,
    previous_discovery_step: str,
    created_at: str,
    *,
    page_cursor: str | None = None,
    raw_extra: dict[str, Any] | None = None,
) -> dict[str, Any]:
    row = {
        "record_schema_version": RUNTIME_RECORD_SCHEMA_VERSION,
        "run_id": run_id,
        "policy_run_id": policy_run_id,
        "search_query_id": search_query_id,
        "search_query": search_query,
        "search_query_generation_method": method,
        "discovery_start_source": discovery_start_source,
        "previous_discovery_step": previous_discovery_step,
        "pattern_seed_ref": {},
        "page_cursor": page_cursor,
        "created_at": created_at,
        **(raw_extra or {}),
    }
    return with_raw_payload(row)


def _walk_action(
    run_id: str,
    policy_run_id: str,
    walk_action_id: str,
    edge_id: str,
    edge_type: str,
    from_node_type: str,
    from_node_id: str,
    to_node_type: str,
    to_node_id: str,
    walk_action: str,
    walk_status: str,
    created_at: str,
    *,
    depth: int = 1,
    page_cursor: str | None = None,
    reason_code: str | None = None,
    budget_tier: str | None = None,
    rule_pack_binding: dict[str, Any] | None = None,
    rule_pack_execution: dict[str, Any] | None = None,
    fallback_rule_pack: dict[str, Any] | None = None,
    raw_extra: dict[str, Any] | None = None,
) -> dict[str, Any]:
    row = with_raw_payload(
        {
            "record_schema_version": RUNTIME_RECORD_SCHEMA_VERSION,
            "run_id": run_id,
            "policy_run_id": policy_run_id,
            "walk_action_id": walk_action_id,
            "edge_id": edge_id,
            "edge_type": edge_type,
            "from_node_type": from_node_type,
            "from_node_id": from_node_id,
            "to_node_type": to_node_type,
            "to_node_id": to_node_id,
            "walk_action": walk_action,
            "walk_status": walk_status,
            "budget_tier": budget_tier or ("normal" if walk_status == "success" else "low_budget"),
            "depth": depth,
            "page_cursor": page_cursor,
            "reason_code": reason_code,
            "rule_pack_id": (rule_pack_binding or {}).get("rule_pack_id")
            or (fallback_rule_pack or {}).get("rule_pack_id"),
            "rule_pack_version": (rule_pack_binding or {}).get("rule_pack_version")
            or (fallback_rule_pack or {}).get("rule_pack_version"),
            "created_at": created_at,
        }
    )
    row["raw_payload"]["rule_pack_binding"] = rule_pack_binding or {}
    row["raw_payload"]["rule_pack_execution"] = rule_pack_execution or {}
    if raw_extra:
        row["raw_payload"].update(raw_extra)
    return row


def _decision_by_content_id(rule_decisions: list[dict[str, Any]]) -> dict[str, dict[str, Any]]:
    return {row["decision_target_id"]: row for row in rule_decisions}


def _edge_permission_for(
    decision: dict[str, Any] | None, edge_id: str, policy: dict[str, Any]
) -> str:
    """判定→边通行证:无判定 / 查询 rule_blocked 一律 deny,其余查 edge_permissions。"""
    if not decision or decision.get("search_query_effect_status") == "rule_blocked":
        return "deny"
    if edge_id in V4_WALK_GATE_EDGES or edge_id == "video_to_hashtag":
        if _is_v4_decision(decision) and not _v4_allow_walk_allowed(decision):
            return "deny"
    return edge_permission(policy, decision.get("decision_action"), edge_id)


def _is_v4_decision(decision: dict[str, Any] | None) -> bool:
    scorecard = (decision or {}).get("scorecard") or {}
    return isinstance(scorecard, dict) and scorecard.get("schema_version") == V4_SCORECARD_SCHEMA_VERSION


def _v4_allow_walk_allowed(decision: dict[str, Any] | None) -> bool:
    if not _is_v4_decision(decision):
        return False
    replay_data = (decision or {}).get("decision_replay_data") or {}
    return replay_data.get("allow_walk") is True


def _v4_walk_gate_denied(decision: dict[str, Any] | None) -> bool:
    return _is_v4_decision(decision) and not _v4_allow_walk_allowed(decision)


def _binding_by_edge_id(walk_strategy: dict[str, Any]) -> dict[str, dict[str, Any]]:
    return {row["edge_id"]: row for row in walk_strategy.get("walk_rule_pack_binding", [])}


def _resolve_edge_binding(
    edge_id: str, walk_strategy: dict[str, Any]
) -> tuple[dict[str, Any], str | None]:
    binding = _binding_by_edge_id(walk_strategy).get(edge_id)
    if not binding:
        return {}, "edge_binding_missing"
    return binding, None


def _execution_record(decision: dict[str, Any] | None, *, content_pack_id: str) -> dict[str, Any]:
    if decision:
        return {
            "executed": True,
            "executed_rule_pack_id": decision.get("rule_pack_id") or content_pack_id,
            "reason": "content_decision_reused_for_walk_gate",
        }
    return {
        "executed": False,
        "executed_rule_pack_id": None,
        "reason": "future_pack_not_enabled",
    }


def _decision_context(decision: dict[str, Any] | None, *, denied: bool = False) -> dict[str, Any]:
    if not decision:
        return {"decision_action": None, "search_query_effect_status": None}
    context = {
        "decision_id": decision.get("decision_id"),
        "decision_action": decision.get("decision_action"),
        "search_query_effect_status": decision.get("search_query_effect_status"),
    }
    if _is_v4_decision(decision):
        context.update(_v4_walk_gate_context(decision, denied=denied))
    return context


def _v4_walk_gate_context(decision: dict[str, Any], *, denied: bool = False) -> dict[str, Any]:
    replay_data = decision.get("decision_replay_data") or {}
    allow_walk = replay_data.get("allow_walk")
    is_denied = denied or allow_walk is not True
    return {
        "allow_walk": allow_walk,
        "allow_walk_reason": replay_data.get("allow_walk_reason"),
        "walk_gate_snapshot": replay_data.get("walk_gate_snapshot"),
        "walk_gate_status": "denied" if is_denied else "allowed",
        "walk_gate_reason_code": V4_ALLOW_WALK_DENIED_REASON if is_denied else None,
    }


def _walk_gate_context_from_query_row(row: dict[str, Any]) -> dict[str, Any]:
    payload = row.get("raw_payload") or {}
    fields = [
        "decision_id",
        "allow_walk",
        "allow_walk_reason",
        "walk_gate_snapshot",
        "walk_gate_status",
        "walk_gate_reason_code",
    ]
    return {field: payload[field] for field in fields if field in payload}


def _blocked_tag(tag: str) -> bool:
    lowered = tag.lower()
    return any(term in lowered for term in BLOCKED_TAG_TERMS)


def _elapsed_ms(start: float, end: float) -> int:
    return max(0, int((end - start) * 1000))


def _walk_action_id(
    run_id: str,
    policy_run_id: str,
    edge_id: str,
    target_id: str,
    suffix: str,
    *,
    depth: int = 1,
) -> str:
    # M8D:并入 depth,防同一 (edge,target,suffix) 跨 frontier 层撞号。
    raw = f"{run_id}:{policy_run_id}:{edge_id}:{target_id}:{suffix}:depth_{depth}"
    return f"wa_{hashlib.sha1(raw.encode('utf-8')).hexdigest()[:16]}"


def _short_hash(value: str) -> str:
    return hashlib.sha1(value.encode("utf-8")).hexdigest()[:10]