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

"""终端边映射与血缘模板(V3-M4 后被 walk_engine._terminal_stage 消费)。

原 run(plan_walk 节点)的编排已并入 walk_engine 终端阶段;本模块只留
决策→终端动作映射与终端边 wa_id(decision_id 后缀,与扩展边的语义后缀是两套约定,冻结)。
"""

from __future__ import annotations

import hashlib
from typing import Any

from content_agent.constants import RUNTIME_RECORD_SCHEMA_VERSION
from content_agent.record_payload import with_raw_payload


def _action_for_decision(decision_action: str) -> dict[str, str]:
    if decision_action == "ADD_TO_CONTENT_POOL":
        return {
            "edge_id": "decision_to_asset",
            "edge_type": "terminal",
            "walk_action": "commit_asset",
            "walk_status": "success",
            "budget_tier": "normal",
        }
    if decision_action == "KEEP_CONTENT_FOR_REVIEW":
        return {
            "edge_id": "budget_downgrade",
            "edge_type": "budget",
            "walk_action": "downgrade_budget",
            "walk_status": "success",
            "budget_tier": "low_budget",
        }
    if decision_action == "TECHNICAL_RETRY_REQUIRED":
        return {
            "edge_id": "path_stop",
            "edge_type": "terminal",
            "walk_action": "stop_path",
            "walk_status": "failed",
            "budget_tier": "stop",
        }
    return {
        "edge_id": "path_stop",
        "edge_type": "terminal",
        "walk_action": "stop_path",
        "walk_status": "skipped",
        "budget_tier": "stop",
    }


def _walk_action_row(
    decision: dict[str, Any],
    item: dict[str, Any],
    decision_action: dict[str, str],
    walk_action_id: str,
    created_at: str,
    binding: dict[str, Any],
    execution: dict[str, Any],
) -> dict[str, Any]:
    row = with_raw_payload(
        {
            "record_schema_version": RUNTIME_RECORD_SCHEMA_VERSION,
            "run_id": decision["run_id"],
            "policy_run_id": decision["policy_run_id"],
            "walk_action_id": walk_action_id,
            "edge_id": decision_action["edge_id"],
            "edge_type": decision_action["edge_type"],
            "from_node_type": "RuleDecision",
            "from_node_id": decision["decision_id"],
            "to_node_type": "Content",
            "to_node_id": item["platform_content_id"],
            "walk_action": decision_action["walk_action"],
            "walk_status": decision_action["walk_status"],
            "budget_tier": decision_action["budget_tier"],
            "depth": 0,
            "page_cursor": item.get("page_cursor"),
            "next_cursor": item.get("next_cursor"),
            "decision_id": decision["decision_id"],
            "rule_pack_id": binding.get("rule_pack_id") or decision["rule_pack_id"],
            "rule_pack_version": binding.get("rule_pack_version") or decision["rule_pack_version"],
            "reason_code": decision["decision_reason_code"],
            "content_effect_status": decision["search_query_effect_status"],
            "decision_target_type": decision.get("decision_target_type"),
            "decision_target_id": decision.get("decision_target_id"),
            "created_at": created_at,
        }
    )
    row["raw_payload"]["rule_pack_binding"] = binding
    row["raw_payload"]["rule_pack_execution"] = execution
    return row


def _walk_action_id(
    run_id: str,
    policy_run_id: str,
    edge_id: str,
    target_id: str,
    decision_id: str,
) -> str:
    raw = f"{run_id}:{policy_run_id}:{edge_id}:{target_id}:{decision_id}"
    return f"wa_{hashlib.sha1(raw.encode('utf-8')).hexdigest()[:16]}"