"""有界游走(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]