feat: add V5 M4 Excel governance tooling

archive/v5_m4_dead_config_archive/20260629T000000Z/douyin_walk_strategy.walk_edge_catalog.json

@@ -0,0 +1,123 @@
+{
+  "source_path": "product_documents/抖音游走策略/douyin_walk_strategy.v1.json#/walk_edge_catalog",
+  "archive_reason": "V5-M4: embedded walk edge catalog compatibility copy retired; canonical source is tech_documents/数据接口与来源/walk_edge_catalog.json.",
+  "walk_edge_catalog": [
+    {
+      "edge_id": "search_page_to_content",
+      "edge_type": "discovery",
+      "from_node_type": "search_page",
+      "to_node_type": "video",
+      "edge_label": "search page to video",
+      "enabled": true,
+      "runtime_stage": "P3/P6",
+      "creates_new_node": true,
+      "can_loop": false,
+      "priority": 20,
+      "notes": "拆出平台视频."
+    },
+    {
+      "edge_id": "video_to_author",
+      "edge_type": "author_seed",
+      "from_node_type": "video",
+      "to_node_type": "author",
+      "edge_label": "video to author",
+      "enabled": true,
+      "runtime_stage": "P6",
+      "creates_new_node": true,
+      "can_loop": false,
+      "priority": 30,
+      "notes": "原 P9 作者一跳入口."
+    },
+    {
+      "edge_id": "author_to_works",
+      "edge_type": "author_works",
+      "from_node_type": "author",
+      "to_node_type": "author_works_page",
+      "edge_label": "author to works",
+      "enabled": true,
+      "runtime_stage": "P6",
+      "creates_new_node": true,
+      "can_loop": true,
+      "priority": 40,
+      "notes": "拉作者作品页."
+    },
+    {
+      "edge_id": "author_work_to_content",
+      "edge_type": "author_work_content",
+      "from_node_type": "author_works_page",
+      "to_node_type": "video",
+      "edge_label": "author work to video",
+      "enabled": true,
+      "runtime_stage": "P6",
+      "creates_new_node": true,
+      "can_loop": false,
+      "priority": 50,
+      "notes": "作者作品重新进入视频判断."
+    },
+    {
+      "edge_id": "video_to_hashtag",
+      "edge_type": "hashtag_seed",
+      "from_node_type": "video",
+      "to_node_type": "hashtag",
+      "edge_label": "video to hashtag",
+      "enabled": true,
+      "runtime_stage": "P6",
+      "creates_new_node": true,
+      "can_loop": false,
+      "priority": 60,
+      "notes": "原 P10 tag 扩散入口."
+    },
+    {
+      "edge_id": "hashtag_to_query",
+      "edge_type": "tag_query",
+      "from_node_type": "hashtag",
+      "to_node_type": "query",
+      "edge_label": "hashtag to query",
+      "enabled": true,
+      "runtime_stage": "P6",
+      "creates_new_node": true,
+      "can_loop": true,
+      "priority": 70,
+      "notes": "强相关 tag 生成 query."
+    },
+    {
+      "edge_id": "decision_to_asset",
+      "edge_type": "asset_commit",
+      "from_node_type": "video",
+      "to_node_type": "asset",
+      "edge_label": "decision to asset",
+      "enabled": true,
+      "runtime_stage": "P7",
+      "creates_new_node": true,
+      "can_loop": false,
+      "priority": 80,
+      "notes": "P5 decision creates final output asset or record."
+    },
+    {
+      "edge_id": "path_stop",
+      "edge_type": "stop",
+      "from_node_type": "video",
+      "to_node_type": "video",
+      "edge_label": "path stop",
+      "enabled": true,
+      "runtime_stage": "P6",
+      "creates_new_node": false,
+      "can_loop": false,
+      "priority": 90,
+      "notes": "Stop current path."
+    },
+    {
+      "edge_id": "budget_downgrade",
+      "edge_type": "budget",
+      "from_node_type": "video",
+      "to_node_type": "video",
+      "edge_label": "budget downgrade",
+      "enabled": true,
+      "runtime_stage": "P6",
+      "creates_new_node": false,
+      "can_loop": false,
+      "priority": 100,
+      "notes": "Lower budget tier."
+    }
+  ]
+}

archive/v5_m4_dead_config_archive/20260629T000000Z/manifest.json

@@ -0,0 +1,19 @@
+{
+  "schema_version": "v5_m4_dead_config_archive_manifest.v1",
+  "archive_batch_id": "v5_m4_dead_config_archive_20260629T000000Z",
+  "items": [
+    {
+      "archive_id": "douyin_walk_strategy_embedded_edge_catalog",
+      "source_path": "product_documents/抖音游走策略/douyin_walk_strategy.v1.json#/walk_edge_catalog",
+      "archive_path": "archive/v5_m4_dead_config_archive/20260629T000000Z/douyin_walk_strategy.walk_edge_catalog.json",
+      "archived_at": "2026-06-29T00:00:00Z",
+      "archived_by": "codex",
+      "proof_no_runtime_read": "V5-M1 moved runtime edge catalog loading to tech_documents/数据接口与来源/walk_edge_catalog.json via walk_graph_json.EDGE_CATALOG_PATH and walk_strategy_json.EDGE_CATALOG_PATH; REQUIRED_SECTIONS no longer includes walk_edge_catalog.",
+      "restore_instruction": "Restore by copying archived walk_edge_catalog back into product_documents/抖音游走策略/douyin_walk_strategy.v1.json before walk_rule_pack_binding.",
+      "stable_until": "",
+      "delete_after": "",
+      "approval_status": "approved",
+      "注释": "M4 only archives this compatibility copy; canonical walk_edge_catalog.json remains active."
+    }
+  ]
+}

scripts/build_config_from_excel.py

@@ -1,4 +1,4 @@
-"""Excel -> runtime JSON converter (V2-M1C), byte-equal via overlay-onto-base.
+"""Excel -> runtime JSON converter (V5-M4), byte-equal via governed generation.
 
 Business edits the config Excel; this regenerates the runtime JSON. To guarantee
 byte-equality (so `policy_bundle_hash` and runtime behaviour never drift on a
@@ -8,6 +8,10 @@ leaves, coercing each cell to the base value's type. Only paths that already
 exist in the base are overlaid — structurally divergent / non-value Excel
 columns are skipped, and non-sheet-backed JSON sections pass through unchanged.
 
+M4 adds governed runtime sheets. Rows from those sheets enter runtime JSON only
+when `approval_status=approved` and `runtime_enabled=true`; other rows are kept
+as Excel ledger/governance data and are intentionally excluded.
+
     python scripts/build_config_from_excel.py --check    # byte-equal? exit 1 on drift
     python scripts/build_config_from_excel.py --write     # regenerate JSON from Excel
 
@@ -18,6 +22,7 @@ changes (adding/removing rows) need a base update — out of V2-M1 scope.
 from __future__ import annotations
 
 import argparse
+import copy
 import json
 import sys
 from pathlib import Path
@@ -36,16 +41,41 @@ RULE_PACK_XLSX = ROOT / "tech_documents/规则包映射/规则包映射配置表
 WALK_XLSX = ROOT / "tech_documents/游走策略/游走策略配置表.xlsx"
 RULE_PACK_JSON = ROOT / "product_documents/规则包/douyin_rule_packs.v1.json"
 WALK_JSON = ROOT / "product_documents/抖音游走策略/douyin_walk_strategy.v1.json"
+SCORE_THRESHOLDS_JSON = ROOT / "tech_documents/数据接口与来源/score_thresholds.json"
+WALK_POLICY_JSON = ROOT / "tech_documents/数据接口与来源/walk_policy.json"
+PROFILE_DIR = ROOT / "tech_documents/数据接口与来源/platform_profiles"
 
 SENTINEL = "备注"  # row 2 of every sheet; skipped
 SKIP_COLUMN = "注释"  # human annotation column with no JSON home
+APPROVAL_STATUSES = {"draft", "waiting_for_human_reivew", "approved", "rejected", "deprecated"}
+RUNTIME_MODES = {"generate_runtime"}
+THRESHOLD_KEYS = {
+    "pool_total",
+    "pool_query",
+    "review_total",
+    "review_query",
+    "walk_query",
+    "walk_platform",
+    "walk_total",
+}
 
 
 # A SheetSpec maps one Excel sheet onto a JSON section by overlay.
 # top-level:  base[section] is a list of objects indexed by id_json (== row[id_excel]).
 # nested:     base["rule_packs"][*][child_path...] grouped by parent_fk (rule_pack_id).
 class SheetSpec:
-    def __init__(self, sheet, *, section, id_excel, id_json=None, renames=None, parent_fk=None, child_path=None):
+    def __init__(
+        self,
+        sheet,
+        *,
+        section=None,
+        id_excel=None,
+        id_json=None,
+        renames=None,
+        parent_fk=None,
+        child_path=None,
+        mode="overlay_existing",
+    ):
         self.sheet = sheet
         self.section = section            # top-level JSON key (for nested: "rule_packs")
         self.id_excel = id_excel
@@ -53,6 +83,7 @@ class SheetSpec:
         self.renames = renames or {}     # excel column -> dotted base path
         self.parent_fk = parent_fk       # excel column holding rule_pack_id (nested only)
         self.child_path = child_path     # dotted path within a rule_pack (nested only)
+        self.mode = mode
 
 
 _WHEN = {"field_path": "when.field", "operator": "when.op", "expected_value": "when.value"}
@@ -82,6 +113,14 @@ _WALK = {
     "walk_fact_contract": "runtime_file",
 }
 WALK_SPECS = [SheetSpec(sheet, section=sheet, id_excel=idc) for sheet, idc in _WALK.items()]
+GOVERNED_RULE_PACK_SHEETS = {
+    "score_weight_profiles": "generate_runtime",
+    "score_threshold_profiles": "generate_runtime",
+}
+GOVERNED_WALK_SHEETS = {
+    "platform_edge_matrix": "generate_runtime",
+    "walk_policy_budget_matrix": "generate_runtime",
+}
 
 
 def _read_rows(ws) -> list[dict[str, Any]]:
@@ -94,6 +133,56 @@ def _read_rows(ws) -> list[dict[str, Any]]:
     return rows
 
 
+def _sheet_rows(wb, sheet: str) -> list[dict[str, Any]]:
+    if sheet not in wb.sheetnames:
+        return []
+    return _read_rows(wb[sheet])
+
+
+def _parse_bool(value: Any) -> bool:
+    if isinstance(value, bool):
+        return value
+    return str(value or "").strip().lower() in {"true", "1", "yes", "y"}
+
+
+def _split_list(value: Any) -> list[str]:
+    if value is None or value == "":
+        return []
+    if isinstance(value, list):
+        return [str(item) for item in value if str(item)]
+    text = str(value).strip()
+    if not text:
+        return []
+    if text.startswith("["):
+        try:
+            parsed = json.loads(text)
+        except json.JSONDecodeError:
+            parsed = []
+        if isinstance(parsed, list):
+            return [str(item) for item in parsed if str(item)]
+    return [part.strip() for part in text.split(",") if part.strip()]
+
+
+def _require_valid_governance(row: dict[str, Any], sheet: str) -> None:
+    status = str(row.get("approval_status") or "").strip()
+    if status not in APPROVAL_STATUSES:
+        raise ValueError(f"{sheet} invalid approval_status: {status!r}")
+    if status == "approved" and _parse_bool(row.get("runtime_enabled")):
+        if not row.get("owner"):
+            raise ValueError(f"{sheet} approved runtime row missing owner")
+        if not row.get("last_reviewed_at"):
+            raise ValueError(f"{sheet} approved runtime row missing last_reviewed_at")
+
+
+def _runtime_row_enabled(row: dict[str, Any], sheet: str) -> bool:
+    _require_valid_governance(row, sheet)
+    return str(row.get("approval_status") or "").strip() == "approved" and _parse_bool(row.get("runtime_enabled"))
+
+
+def _load_workbook(path: Path, *, read_only: bool = True):
+    return load_workbook(path, data_only=True, read_only=read_only)
+
+
 def _coerce(cell: Any, base_value: Any) -> Any:
     if isinstance(base_value, bool):
         return str(cell).strip().lower() in {"true", "1", "yes"} if not isinstance(cell, bool) else cell
@@ -145,8 +234,10 @@ def _nested_targets(base: dict[str, Any], spec: SheetSpec) -> dict[tuple, dict[s
 
 
 def overlay_workbook(xlsx: Path, base: dict[str, Any], specs: list[SheetSpec]) -> dict[str, Any]:
-    wb = load_workbook(xlsx, data_only=True, read_only=True)
+    wb = _load_workbook(xlsx)
     for spec in specs:
+        if spec.mode in RUNTIME_MODES:
+            raise ValueError(f"{spec.sheet} is a runtime sheet and needs a dedicated builder")
         rows = _read_rows(wb[spec.sheet])
         if spec.child_path:
             index = _nested_targets(base, spec)
@@ -163,6 +254,175 @@ def overlay_workbook(xlsx: Path, base: dict[str, Any], specs: list[SheetSpec]) -
     return base
 
 
+def _build_score_weight_profiles(rows: list[dict[str, Any]]) -> dict[str, Any] | None:
+    runtime_rows = [row for row in rows if _runtime_row_enabled(row, "score_weight_profiles")]
+    if not runtime_rows:
+        return None
+    profiles: dict[str, dict[str, Any]] = {}
+    for row in runtime_rows:
+        profile_id = str(row.get("profile_id") or "").strip()
+        dimension_key = str(row.get("dimension_key") or "").strip()
+        if not profile_id or not dimension_key:
+            raise ValueError("score_weight_profiles runtime rows require profile_id and dimension_key")
+        profile = profiles.setdefault(
+            profile_id,
+            {
+                "profile_id": profile_id,
+                "profile_label": row.get("profile_label") or profile_id,
+                "historical_alias_ids": _split_list(row.get("historical_alias_ids")),
+                "platform_scope": _split_list(row.get("platform_scope")) or ["*"],
+                "content_format": row.get("content_format") or "*",
+                "applies_when": row.get("applies_when") or "always",
+                "weights": {},
+                "normalization_policy": row.get("normalization_policy") or "sum_100",
+                "normalization_reason": row.get("normalization_reason") or "",
+                "missing_dimension_policy": row.get("missing_dimension_policy") or "score_missing",
+                "critical_dimensions": [],
+                "score_source_paths": {},
+                "priority": int(row.get("priority") or 0),
+                "fallback_profile_id": row.get("fallback_profile_id") or None,
+                "runtime_enabled": True,
+                "approval_status": "approved",
+                "owner": row.get("owner"),
+                "last_reviewed_at": str(row.get("last_reviewed_at")),
+            },
+        )
+        profile["weights"][dimension_key] = _number(row.get("weight_percent"))
+        score_source_path = row.get("score_source_path")
+        if score_source_path:
+            profile["score_source_paths"][dimension_key] = str(score_source_path)
+        if _parse_bool(row.get("critical_dimension")):
+            profile["critical_dimensions"].append(dimension_key)
+    return {
+        "schema_version": "score_weight_profiles.v2",
+        "profiles": list(profiles.values()),
+    }
+
+
+def _apply_score_weight_profiles(rule_pack: dict[str, Any], rows: list[dict[str, Any]]) -> None:
+    built = _build_score_weight_profiles(rows)
+    if built is None:
+        return
+    for pack in rule_pack.get("rule_packs", []):
+        scorecard = pack.get("scorecard")
+        if isinstance(scorecard, dict):
+            scorecard["score_weight_profiles"] = built
+
+
+def _build_score_thresholds(base: dict[str, Any], rows: list[dict[str, Any]]) -> dict[str, Any]:
+    runtime_rows = [row for row in rows if _runtime_row_enabled(row, "score_threshold_profiles")]
+    if not runtime_rows:
+        return base
+    thresholds = {
+        key: copy.deepcopy(value)
+        for key, value in base.items()
+        if key.startswith("_") or key == "schema_version"
+    }
+    for row in runtime_rows:
+        key = str(row.get("threshold_key") or "").strip()
+        if key not in THRESHOLD_KEYS:
+            raise ValueError(f"score_threshold_profiles unknown threshold_key: {key}")
+        value = row.get("threshold_value")
+        if value is None or value == "":
+            raise ValueError(f"score_threshold_profiles missing value for {key}")
+        scope_values = _split_list(row.get("platform_scope")) or ["default"]
+        for scope in scope_values:
+            target = "default" if scope == "*" else scope
+            thresholds.setdefault(target, {})[key] = int(round(float(value)))
+    return thresholds
+
+
+def _build_platform_profiles(rows: list[dict[str, Any]]) -> dict[str, dict[str, Any]]:
+    by_platform: dict[str, dict[str, Any]] = {}
+    for profile_path in PROFILE_DIR.glob("*.json"):
+        profile, _ = config_store.load_json(profile_path)
+        by_platform[profile_path.stem] = profile
+    for row in rows:
+        if not _runtime_row_enabled(row, "platform_edge_matrix"):
+            continue
+        platform = str(row.get("platform") or "").strip()
+        edge_id = str(row.get("edge_id") or "").strip()
+        status = str(row.get("status") or "").strip()
+        if platform not in by_platform:
+            raise ValueError(f"platform_edge_matrix unknown platform: {platform}")
+        if not edge_id or not status:
+            raise ValueError("platform_edge_matrix runtime rows require platform, edge_id, status")
+        spec = by_platform[platform].setdefault("edges", {}).setdefault(edge_id, {})
+        spec["status"] = status
+        if row.get("status_reason"):
+            spec["reason"] = str(row.get("status_reason"))
+    return by_platform
+
+
+def _set_wrapped_value(container: dict[str, Any], key: str, value: Any) -> None:
+    current = container.get(key)
+    if isinstance(current, dict) and "value" in current:
+        current["value"] = value
+    else:
+        container[key] = value
+
+
+def _build_walk_policy(base: dict[str, Any], rows: list[dict[str, Any]]) -> dict[str, Any]:
+    runtime_rows = [row for row in rows if _runtime_row_enabled(row, "walk_policy_budget_matrix")]
+    if not runtime_rows:
+        return base
+    policy = copy.deepcopy(base)
+    platform_rows = []
+    for row in runtime_rows:
+        platform = str(row.get("platform") or "").strip()
+        edge_id = str(row.get("edge_id") or "").strip()
+        if not platform or not edge_id:
+            raise ValueError("walk_policy_budget_matrix runtime rows require platform and edge_id")
+        if platform == "default" and edge_id == "__global__":
+            platform_rows.append(
+                {
+                    "platform": platform,
+                    "edge_id": edge_id,
+                    "max_total_actions": _nullable_int(row.get("max_total_actions")),
+                    "max_depth": _nullable_int(row.get("max_depth")),
+                    "budget_exhausted_reason_code": row.get("budget_exhausted_reason_code") or "budget_exhausted",
+                    "operator_display_status": row.get("operator_display_status") or "budget_blocked",
+                }
+            )
+            if row.get("max_total_actions") not in {None, ""}:
+                _set_wrapped_value(policy["global"], "max_total_actions_per_run", int(row.get("max_total_actions")))
+            if row.get("max_depth") not in {None, ""}:
+                _set_wrapped_value(policy["global"], "max_depth", int(row.get("max_depth")))
+            continue
+        budget_row = {
+            "platform": platform,
+            "edge_id": edge_id,
+            "max_total_actions": _nullable_int(row.get("max_total_actions")),
+            "max_per_content": _nullable_int(row.get("max_per_content")),
+            "max_works_per_author": _nullable_int(row.get("max_works_per_author")),
+            "max_pages": _nullable_int(row.get("max_pages")),
+            "budget_exhausted_reason_code": row.get("budget_exhausted_reason_code") or "budget_exhausted",
+            "operator_display_status": row.get("operator_display_status") or "budget_blocked",
+        }
+        platform_rows.append(budget_row)
+        if platform == "default":
+            legacy = next((item for item in policy.get("edge_budgets", []) if item.get("edge_id") == edge_id), None)
+            if legacy is None:
+                legacy = {"edge_id": edge_id}
+                policy.setdefault("edge_budgets", []).append(legacy)
+            for field in ["max_total_actions", "max_per_content", "max_works_per_author", "max_pages"]:
+                if budget_row[field] is not None:
+                    legacy[field] = budget_row[field]
+    policy["platform_edge_budgets"] = platform_rows
+    return policy
+
+
+def _nullable_int(value: Any) -> int | None:
+    if value is None or value == "":
+        return None
+    return int(round(float(value)))
+
+
+def _number(value: Any) -> int | float:
+    parsed = float(value)
+    return int(parsed) if parsed.is_integer() else parsed
+
+
 def _get_path(obj: dict[str, Any], dotted: str) -> tuple[bool, Any]:
     cursor: Any = obj
     for part in dotted.split("."):
@@ -186,7 +446,7 @@ def sync_workbook(xlsx: Path, base: dict[str, Any], specs: list[SheetSpec]) -> i
     Used to reconcile a drifted Excel to the authoritative JSON (JSON wins). Only
     mapped value cells are overwritten; id / FK / 注释 / unmapped columns are left.
     """
-    wb = load_workbook(xlsx)
+    wb = _load_workbook(xlsx, read_only=False)
     changed = 0
     for spec in specs:
         ws = wb[spec.sheet]
@@ -222,13 +482,51 @@ def sync_workbook(xlsx: Path, base: dict[str, Any], specs: list[SheetSpec]) -> i
 
 
 def build() -> dict[str, dict[str, Any]]:
-    """Return {json_path: regenerated_obj} for both config files."""
+    """Return {json_path: regenerated_obj} for all runtime config files."""
     rule_pack_base, _ = config_store.load_json(RULE_PACK_JSON)
     walk_base, _ = config_store.load_json(WALK_JSON)
-    return {
-        str(RULE_PACK_JSON): overlay_workbook(RULE_PACK_XLSX, rule_pack_base, RULE_PACK_SPECS),
-        str(WALK_JSON): overlay_workbook(WALK_XLSX, walk_base, WALK_SPECS),
+    score_threshold_base, _ = config_store.load_json(SCORE_THRESHOLDS_JSON)
+    walk_policy_base, _ = config_store.load_json(WALK_POLICY_JSON)
+    rule_wb = _load_workbook(RULE_PACK_XLSX)
+    walk_wb = _load_workbook(WALK_XLSX)
+    rule_pack = overlay_workbook(RULE_PACK_XLSX, rule_pack_base, RULE_PACK_SPECS)
+    _apply_score_weight_profiles(rule_pack, _sheet_rows(rule_wb, "score_weight_profiles"))
+    score_thresholds = _build_score_thresholds(score_threshold_base, _sheet_rows(rule_wb, "score_threshold_profiles"))
+    walk_strategy = overlay_workbook(WALK_XLSX, walk_base, WALK_SPECS)
+    walk_strategy.pop("walk_edge_catalog", None)
+    walk_policy = _build_walk_policy(walk_policy_base, _sheet_rows(walk_wb, "walk_policy_budget_matrix"))
+    built = {
+        str(RULE_PACK_JSON): rule_pack,
+        str(WALK_JSON): walk_strategy,
+        str(SCORE_THRESHOLDS_JSON): score_thresholds,
+        str(WALK_POLICY_JSON): walk_policy,
     }
+    for platform, profile in _build_platform_profiles(_sheet_rows(walk_wb, "platform_edge_matrix")).items():
+        built[str(PROFILE_DIR / f"{platform}.json")] = profile
+    return built
+
+
+def governance_report() -> list[dict[str, Any]]:
+    """Summarize governed sheets so no-op sheets and runtime rows are visible in gates."""
+    report: list[dict[str, Any]] = []
+    for xlsx, sheets in [
+        (RULE_PACK_XLSX, GOVERNED_RULE_PACK_SHEETS),
+        (WALK_XLSX, GOVERNED_WALK_SHEETS),
+    ]:
+        wb = _load_workbook(xlsx)
+        for sheet, mode in sheets.items():
+            rows = _sheet_rows(wb, sheet)
+            runtime_count = sum(1 for row in rows if _runtime_row_enabled(row, sheet))
+            report.append(
+                {
+                    "workbook": str(xlsx.relative_to(ROOT)),
+                    "sheet": sheet,
+                    "mode": mode,
+                    "row_count": len(rows),
+                    "runtime_row_count": runtime_count,
+                }
+            )
+    return report
 
 
 def _first_diff(expected: str, actual: str) -> str:
@@ -266,7 +564,7 @@ def main() -> int:
                 entry["first_diff"] = _first_diff(current, generated)
             findings.append(entry)
     status = "written" if args.write else ("pass" if all(f.get("byte_equal") for f in findings) else "fail")
-    print(json.dumps({"status": status, "findings": findings}, ensure_ascii=False, indent=2))
+    print(json.dumps({"status": status, "findings": findings, "governance": governance_report()}, ensure_ascii=False, indent=2))
     return 0 if status in {"written", "pass"} else 1
 
 

scripts/run_config_gate.py

@@ -23,6 +23,7 @@ GATES = [
     ("excel_json_sync", ["scripts/validate_config_excel_sync.py"]),
     ("query_prompts", ["scripts/validate_query_prompts_config.py"]),
     ("v4_config_contract", ["scripts/validate_v4_config_contract.py"]),
+    ("v5_m4_dead_config_archive", ["scripts/validate_v5_m4_dead_config_archive.py"]),
 ]
 
 

scripts/validate_config_excel_sync.py

@@ -16,7 +16,7 @@ ROOT = Path(__file__).resolve().parents[1]
 if str(ROOT) not in sys.path:
     sys.path.insert(0, str(ROOT))
 
-from scripts.build_config_from_excel import build
+from scripts.build_config_from_excel import build, governance_report
 from scripts.check_config_json_canonical import canonical_dumps
 
 
@@ -30,6 +30,19 @@ def main() -> int:
         if not ok:
             finding["message"] = "Excel diverges from JSON; run build_config_from_excel.py --sync-excel (JSON wins) or --write (Excel wins)"
         findings.append(finding)
+    governance = governance_report()
+    for row in governance:
+        findings.append(
+            {
+                "level": "info",
+                "check_id": "excel_governance_sheet",
+                "config_path": row["workbook"],
+                "sheet": row["sheet"],
+                "mode": row["mode"],
+                "row_count": row["row_count"],
+                "runtime_row_count": row["runtime_row_count"],
+            }
+        )
     status = "fail" if any(f["level"] == "fail" for f in findings) else "pass"
     print(json.dumps({"status": status, "findings": findings}, ensure_ascii=False, indent=2))
     return 1 if status == "fail" else 0

scripts/validate_rule_pack_config.py

@@ -23,6 +23,17 @@ LEGACY_FIELD_BLOCKLIST = {
     "platform_heat",
     "age_50_plus_level",
 }
+V4_BASE_ACTIVE_DIMENSIONS = {"query_relevance", "platform_performance"}
+V4_REQUIRED_SCORE_WEIGHT_PROFILES = {
+    "default_two_dimension",
+    "douyin_fifty_plus_ok",
+    "douyin_fifty_plus_not_attempted",
+}
+V4_SCORE_WEIGHT_ALIAS_TARGETS = {
+    "base_video_two_dimension_v1": "default_two_dimension",
+    "audience_50plus_available_v1": "douyin_fifty_plus_ok",
+    "audience_50plus_skipped_by_query_gate_v1": "douyin_fifty_plus_not_attempted",
+}
 
 
 def _fail(findings: list[dict[str, Any]], check_id: str, message: str) -> None:
@@ -94,13 +105,8 @@ def _check_v4_rule_pack(findings: list[dict[str, Any]], pack: dict[str, Any]) ->
     scorecard = pack.get("scorecard") or {}
     if scorecard.get("schema_version") != "v4_scorecard.v1":
         _fail(findings, "v4_scorecard_schema", f"{pid} scorecard.schema_version must be v4_scorecard.v1")
-    dimensions = [row for row in scorecard.get("dimensions", []) if row.get("runtime_status") == "active"]
-    keys = [row.get("key") for row in dimensions]
-    if keys != ["query_relevance", "platform_performance"]:
-        _fail(findings, "v4_scorecard_dimensions", f"{pid} active dimensions must be query_relevance/platform_performance, got {keys}")
-    for row in dimensions:
-        if row.get("max_score") != 50 or row.get("weight_percent") != 50:
-            _fail(findings, "v4_scorecard_weight", f"{pid}/{row.get('key')} must be max_score=50 weight_percent=50")
+    _check_scorecard_dimensions(findings, pid, scorecard)
+    _check_score_weight_profiles(findings, pid, scorecard)
     thresholds = pack.get("thresholds", [])
     expected_reasons = {
         "v4_query_and_platform_pass",
@@ -137,6 +143,141 @@ def _legacy_paths(value: Any, prefix: str) -> list[str]:
     return paths
 
 
+def _check_scorecard_dimensions(
+    findings: list[dict[str, Any]],
+    pid: Any,
+    scorecard: dict[str, Any],
+) -> None:
+    dimensions = scorecard.get("dimensions")
+    if not isinstance(dimensions, list):
+        _fail(findings, "v4_scorecard_dimensions", f"{pid} scorecard.dimensions must be a list")
+        return
+    keys: list[Any] = []
+    for row in dimensions:
+        if not isinstance(row, dict):
+            _fail(findings, "v4_scorecard_dimension", f"{pid} dimension row must be object")
+            continue
+        key = row.get("key")
+        if not key:
+            _fail(findings, "v4_scorecard_dimension_key", f"{pid} dimension key must be non-empty")
+        keys.append(key)
+        if row.get("runtime_status") != "active":
+            continue
+        active = row.get("active")
+        if active is not None and active is not True:
+            _fail(findings, "v4_scorecard_dimension_active", f"{pid}/{key} active must match runtime_status=active")
+        platform_scope = row.get("platform_scope")
+        if platform_scope is not None and (
+            not isinstance(platform_scope, list)
+            or not platform_scope
+            or any(not isinstance(item, str) or not item for item in platform_scope)
+        ):
+            _fail(findings, "v4_scorecard_dimension_platform_scope", f"{pid}/{key} platform_scope must be a non-empty string list")
+        score_source_path = row.get("score_source_path")
+        if score_source_path is not None and (not isinstance(score_source_path, str) or not score_source_path):
+            _fail(findings, "v4_scorecard_dimension_score_source_path", f"{pid}/{key} score_source_path must be non-empty string")
+        if not _positive_number(row.get("max_score")):
+            _fail(findings, "v4_scorecard_dimension_score", f"{pid}/{key} max_score must be positive")
+        if not _positive_number(row.get("weight_percent")):
+            _fail(findings, "v4_scorecard_dimension_weight", f"{pid}/{key} weight_percent must be positive")
+    duplicate_keys = sorted({key for key in keys if key and keys.count(key) > 1})
+    if duplicate_keys:
+        _fail(findings, "v4_scorecard_dimension_duplicate", f"{pid} duplicate dimensions: {duplicate_keys}")
+    active_keys = {row.get("key") for row in dimensions if isinstance(row, dict) and row.get("runtime_status") == "active"}
+    missing = sorted(V4_BASE_ACTIVE_DIMENSIONS - active_keys)
+    if missing:
+        _fail(findings, "v4_scorecard_dimensions", f"{pid} active dimensions missing base keys: {missing}")
+
+
+def _check_score_weight_profiles(
+    findings: list[dict[str, Any]],
+    pid: Any,
+    scorecard: dict[str, Any],
+) -> None:
+    profiles = scorecard.get("score_weight_profiles")
+    if not isinstance(profiles, dict):
+        _fail(findings, "v4_score_weight_profiles", f"{pid} score_weight_profiles must be an object")
+        return
+    dimension_keys = {
+        row.get("key")
+        for row in scorecard.get("dimensions", [])
+        if isinstance(row, dict) and row.get("key")
+    }
+    if isinstance(profiles.get("profiles"), list):
+        _check_generic_score_weight_profiles(findings, pid, profiles["profiles"], dimension_keys)
+        return
+    missing_profiles = sorted(V4_REQUIRED_SCORE_WEIGHT_PROFILES - set(profiles))
+    if missing_profiles:
+        _fail(findings, "v4_score_weight_profiles", f"{pid} missing score weight profiles: {missing_profiles}")
+    for profile_name, weights in profiles.items():
+        if not isinstance(weights, dict) or not weights:
+            _fail(findings, "v4_score_weight_profile", f"{pid}/{profile_name} must be a non-empty object")
+            continue
+        for key, weight in weights.items():
+            if key not in dimension_keys:
+                _fail(findings, "v4_score_weight_profile_dimension", f"{pid}/{profile_name} unknown dimension: {key}")
+            if not _positive_number(weight):
+                _fail(findings, "v4_score_weight_profile_weight", f"{pid}/{profile_name}/{key} weight must be positive")
+
+
+def _check_generic_score_weight_profiles(
+    findings: list[dict[str, Any]],
+    pid: Any,
+    profiles: list[Any],
+    dimension_keys: set[Any],
+) -> None:
+    aliases: set[str] = set()
+    priority_groups: dict[tuple[Any, Any, Any, int], list[str]] = {}
+    for profile in profiles:
+        if not isinstance(profile, dict):
+            _fail(findings, "v4_score_weight_profile", f"{pid} profile row must be object")
+            continue
+        profile_id = profile.get("profile_id")
+        if not profile_id:
+            _fail(findings, "v4_score_weight_profile", f"{pid} profile_id must be non-empty")
+            continue
+        aliases.update(profile.get("historical_alias_ids") or [])
+        if profile.get("approval_status") != "approved":
+            _fail(findings, "v4_score_weight_profile_approval", f"{pid}/{profile_id} runtime profile must be approved")
+        if profile.get("runtime_enabled") is not True:
+            _fail(findings, "v4_score_weight_profile_runtime", f"{pid}/{profile_id} runtime_enabled must be true")
+        weights = profile.get("weights")
+        if not isinstance(weights, dict) or not weights:
+            _fail(findings, "v4_score_weight_profile", f"{pid}/{profile_id} weights must be non-empty object")
+            continue
+        total = 0.0
+        for key, weight in weights.items():
+            if key not in dimension_keys:
+                _fail(findings, "v4_score_weight_profile_dimension", f"{pid}/{profile_id} unknown dimension: {key}")
+            if not _positive_number(weight):
+                _fail(findings, "v4_score_weight_profile_weight", f"{pid}/{profile_id}/{key} weight must be positive")
+            else:
+                total += float(weight)
+        normalization = profile.get("normalization_policy") or "sum_100"
+        if normalization == "raw_sum":
+            if not profile.get("normalization_reason"):
+                _fail(findings, "v4_score_weight_profile_normalization", f"{pid}/{profile_id} raw_sum requires normalization_reason")
+        elif round(total, 6) != 100:
+            _fail(findings, "v4_score_weight_profile_total", f"{pid}/{profile_id} weights must sum to 100")
+        group = (
+            tuple(profile.get("platform_scope") or []),
+            profile.get("content_format") or "*",
+            profile.get("applies_when") or "always",
+            int(profile.get("priority") or 0),
+        )
+        priority_groups.setdefault(group, []).append(str(profile_id))
+    missing = sorted(V4_REQUIRED_SCORE_WEIGHT_PROFILES - aliases)
+    if missing:
+        _fail(findings, "v4_score_weight_profiles", f"{pid} missing historical aliases: {missing}")
+    for group, ids in priority_groups.items():
+        if len(ids) > 1:
+            _fail(findings, "v4_score_weight_profile_priority", f"{pid} priority conflict {group}: {ids}")
+
+
+def _positive_number(value: Any) -> bool:
+    return isinstance(value, (int, float)) and not isinstance(value, bool) and value > 0
+
+
 def _string_has_legacy_field(value: str) -> bool:
     normalized = value.replace("[", ".").replace("]", ".").replace("/", ".")
     parts = [part.strip() for part in normalized.split(".")]

scripts/validate_v4_config_contract.py

@@ -12,6 +12,7 @@ ROOT = Path(__file__).resolve().parents[1]
 DATA_DIR = ROOT / "tech_documents/数据接口与来源"
 RULE_PACK_PATH = ROOT / "product_documents/规则包/douyin_rule_packs.v1.json"
 WALK_STRATEGY_PATH = ROOT / "product_documents/抖音游走策略/douyin_walk_strategy.v1.json"
+WALK_EDGE_CATALOG_PATH = ROOT / "tech_documents/数据接口与来源/walk_edge_catalog.json"
 LEGACY_FIELD_BLOCKLIST = {
     "fit_senior_50plus",
     "fit_confidence",
@@ -27,7 +28,7 @@ ENDPOINT_STATUSES = {
     "missing",
 }
 M2_PLATFORM_PROFILES = {"douyin", "kuaishou", "shipinhao"}
-PROFILE_EDGE_STATUSES = {"supported", "blocked"}
+PROFILE_EDGE_STATUSES = {"supported", "blocked", "no_interface", "planned"}
 PROFILE_ENDPOINT_STATUSES = {"verified", "verified_unstable", "blocked", "source_only", "missing"}
 OBSERVABLE_FIELDS = {
     "statistics.digg_count",
@@ -39,6 +40,12 @@ OBSERVABLE_FIELDS = {
 MISSING_OBSERVABLE_TYPES = {"natural_platform_missing", "runtime_missing"}
 M4_WALK_GATE_EDGES = {"hashtag_to_query", "author_to_works"}
 M4_WALK_GATE_RAW_FIELDS = {"decision_id", "allow_walk", "allow_walk_reason", "walk_gate_snapshot"}
+V4_BASE_ACTIVE_DIMENSIONS = {"query_relevance", "platform_performance"}
+V4_REQUIRED_SCORE_WEIGHT_PROFILES = {
+    "default_two_dimension",
+    "douyin_fifty_plus_ok",
+    "douyin_fifty_plus_not_attempted",
+}
 
 
 def main() -> int:
@@ -53,11 +60,41 @@ def validate_v4_config_contract(root: Path = ROOT) -> list[dict[str, str]]:
     findings: list[dict[str, str]] = []
 
     walk_graph = _load_json(data_dir / "walk_graph.json", findings)
+    graph_edges: set[str] = set()
     if walk_graph:
         _check_no_legacy_fields(findings, walk_graph, "walk_graph.json")
         _check_value(findings, "walk_graph_schema", walk_graph.get("schema_version"), "walk_graph.v2")
         _check_count(findings, "walk_graph_nodes", "walk_graph.nodes", walk_graph.get("nodes"), 8)
         _check_count(findings, "walk_graph_edges", "walk_graph.edges", walk_graph.get("edges"), 9)
+        graph_edges = {
+            edge.get("edge_id")
+            for edge in walk_graph.get("edges", [])
+            if isinstance(edge, dict) and edge.get("edge_id")
+        }
+
+    walk_edge_catalog = _load_json(data_dir / "walk_edge_catalog.json", findings)
+    catalog_edges: set[str] = set()
+    if walk_edge_catalog:
+        _check_no_legacy_fields(findings, walk_edge_catalog, "walk_edge_catalog.json")
+        _check_value(
+            findings,
+            "walk_edge_catalog_schema",
+            walk_edge_catalog.get("schema_version"),
+            "walk_edge_catalog.v1",
+        )
+        rows = walk_edge_catalog.get("walk_edge_catalog")
+        _check_count(findings, "walk_edge_catalog_edges", "walk_edge_catalog", rows, 9)
+        catalog_edges = {
+            row.get("edge_id")
+            for row in rows or []
+            if isinstance(row, dict) and row.get("edge_id")
+        }
+        if walk_graph and graph_edges != catalog_edges:
+            _fail(
+                findings,
+                "walk_graph_catalog_mismatch",
+                f"walk_graph edges and walk_edge_catalog differ: graph_only={sorted(graph_edges - catalog_edges)}, catalog_only={sorted(catalog_edges - graph_edges)}",
+            )
 
     walk_policy = _load_json(data_dir / "walk_policy.json", findings)
     if walk_policy:
@@ -90,7 +127,7 @@ def validate_v4_config_contract(root: Path = ROOT) -> list[dict[str, str]]:
         if not isinstance(endpoints, list):
             _fail(findings, "crawler_endpoints_invalid", "endpoints must be a list")
         else:
-            _check_count(findings, "crawler_endpoints_count", "endpoints", endpoints, 26)
+            _check_count(findings, "crawler_endpoints_count", "endpoints", endpoints, 27)
             for endpoint in endpoints:
                 if not isinstance(endpoint, dict):
                     _fail(findings, "crawler_endpoint_invalid", "endpoint row must be object")
@@ -153,6 +190,7 @@ def validate_v4_config_contract(root: Path = ROOT) -> list[dict[str, str]]:
         for key in ["platform", "status", "runtime", "endpoints", "edges"]:
             if key not in profile:
                 _fail(findings, "platform_profile_missing_key", f"{profile_path.name} missing {key}")
+        _check_platform_profile_edges(findings, profile, profile_path.name, graph_edges or catalog_edges)
         if profile.get("platform") in M2_PLATFORM_PROFILES:
             _check_m2_platform_profile(findings, profile, profile_path.name)
 
@@ -163,7 +201,7 @@ def validate_v4_config_contract(root: Path = ROOT) -> list[dict[str, str]]:
     walk_strategy = _load_json(WALK_STRATEGY_PATH, findings)
     if walk_strategy:
         _check_no_legacy_fields(findings, walk_strategy, "douyin_walk_strategy.v1.json")
-        _check_v4_walk_strategy_contract(findings, walk_strategy)
+        _check_v4_walk_strategy_contract(findings, walk_strategy, catalog_edges or graph_edges)
 
     return findings
 
@@ -244,27 +282,37 @@ def _check_count(
         _fail(findings, check_id, f"{label} expected {expected}, got {actual}")
 
 
-def _check_m2_platform_profile(
+def _check_platform_profile_edges(
     findings: list[dict[str, str]],
     profile: dict[str, Any],
     label: str,
+    edge_ids: set[str],
 ) -> None:
-    platform = str(profile.get("platform") or "")
     edges = profile.get("edges")
     if not isinstance(edges, dict):
         _fail(findings, "platform_profile_edges_invalid", f"{label} edges must be an object")
-    else:
-        for edge_id, edge in edges.items():
-            if not isinstance(edge, dict):
-                _fail(findings, "platform_profile_edge_invalid", f"{label}.{edge_id} must be object")
-                continue
-            status = edge.get("status")
-            if status not in PROFILE_EDGE_STATUSES:
-                _fail(
-                    findings,
-                    "platform_profile_edge_status_unknown",
-                    f"{label}.{edge_id} status must be supported/blocked, got {status}",
-                )
+        return
+    for edge_id, edge in edges.items():
+        if edge_ids and edge_id not in edge_ids:
+            _fail(findings, "platform_profile_edge_unknown", f"{label}.{edge_id} unknown edge")
+        if not isinstance(edge, dict):
+            _fail(findings, "platform_profile_edge_invalid", f"{label}.{edge_id} must be object")
+            continue
+        status = edge.get("status")
+        if status not in PROFILE_EDGE_STATUSES:
+            _fail(
+                findings,
+                "platform_profile_edge_status_unknown",
+                f"{label}.{edge_id} status must be supported/blocked/no_interface/planned, got {status}",
+            )
+
+
+def _check_m2_platform_profile(
+    findings: list[dict[str, str]],
+    profile: dict[str, Any],
+    label: str,
+) -> None:
+    platform = str(profile.get("platform") or "")
 
     endpoints = profile.get("endpoints")
     if not isinstance(endpoints, dict):
@@ -374,8 +422,8 @@ def _check_m2_platform_specifics(
     endpoints = profile.get("endpoints") if isinstance(profile.get("endpoints"), dict) else {}
 
     if platform == "kuaishou":
-        _check_nested_status(findings, label, edges, "author_to_works", "blocked")
-        _check_nested_status(findings, label, edges, "author_work_to_content", "blocked")
+        _check_nested_status(findings, label, edges, "author_to_works", "supported")
+        _check_nested_status(findings, label, edges, "author_work_to_content", "supported")
     if platform == "shipinhao":
         _check_nested_status(findings, label, endpoints, "account_info", "blocked")
         _check_nested_status(findings, label, edges, "author_to_works", "blocked")
@@ -406,7 +454,15 @@ def _check_v4_walk_gate_config(
 def _check_v4_walk_strategy_contract(
     findings: list[dict[str, str]],
     strategy: dict[str, Any],
+    edge_ids: set[str],
 ) -> None:
+    for binding in strategy.get("walk_rule_pack_binding") or []:
+        if isinstance(binding, dict) and edge_ids and binding.get("edge_id") not in edge_ids:
+            _fail(
+                findings,
+                "v4_walk_strategy_edge_unknown",
+                f"walk_rule_pack_binding unknown edge_id: {binding.get('edge_id')}",
+            )
     rows = strategy.get("v4_walk_gate")
     if not isinstance(rows, list) or not rows:
         _fail(findings, "v4_walk_strategy_gate_missing", "douyin_walk_strategy.v1.json missing v4_walk_gate")
@@ -417,6 +473,9 @@ def _check_v4_walk_strategy_contract(
         _fail(findings, "v4_walk_strategy_gate_missing", "allow_walk_required gate missing")
         return
     _check_v4_walk_gate_config(findings, gate, "douyin_walk_strategy.v4_walk_gate.allow_walk_required")
+    for edge_id in gate.get("applies_to_edges") or []:
+        if edge_ids and edge_id not in edge_ids:
+            _fail(findings, "v4_walk_strategy_gate_edge_unknown", f"v4_walk_gate unknown edge_id: {edge_id}")
 
 
 def _check_nested_status(
@@ -465,14 +524,9 @@ def _check_v4_rule_pack_contract(
                 "v4_scorecard_schema_invalid",
                 f"{pack.get('rule_pack_id')} scorecard.schema_version must be v4_scorecard.v1",
             )
-        dimensions = [row for row in scorecard.get("dimensions", []) if row.get("runtime_status") == "active"]
-        keys = [row.get("key") for row in dimensions]
-        if keys != ["query_relevance", "platform_performance"]:
-            _fail(
-                findings,
-                "v4_scorecard_dimensions_invalid",
-                f"{pack.get('rule_pack_id')} active dimensions must be query_relevance/platform_performance",
-            )
+        _check_v4_scorecard_dimensions(findings, pack.get("rule_pack_id"), scorecard)
+        _check_v4_score_weight_profiles(findings, pack.get("rule_pack_id"), scorecard)
+        _check_v4_scorecard_rule_refs(findings, pack.get("rule_pack_id"), scorecard)
         required_fields = set((pack.get("input_contract") or {}).get("required_fields") or [])
         for field in [
             "pattern_match_result.query_relevance_score",
@@ -487,6 +541,199 @@ def _check_v4_rule_pack_contract(
                 )
 
 
+def _check_v4_scorecard_dimensions(
+    findings: list[dict[str, str]],
+    rule_pack_id: Any,
+    scorecard: dict[str, Any],
+) -> None:
+    dimensions = scorecard.get("dimensions")
+    if not isinstance(dimensions, list):
+        _fail(findings, "v4_scorecard_dimensions_invalid", f"{rule_pack_id} dimensions must be a list")
+        return
+    keys: list[Any] = []
+    for row in dimensions:
+        if not isinstance(row, dict):
+            _fail(findings, "v4_scorecard_dimension_invalid", f"{rule_pack_id} dimension row must be object")
+            continue
+        key = row.get("key")
+        if not key:
+            _fail(findings, "v4_scorecard_dimension_key_invalid", f"{rule_pack_id} dimension key must be non-empty")
+        keys.append(key)
+        if row.get("runtime_status") != "active":
+            continue
+        active = row.get("active")
+        if active is not None and active is not True:
+            _fail(
+                findings,
+                "v4_scorecard_dimension_active_invalid",
+                f"{rule_pack_id}/{key} active must match runtime_status=active",
+            )
+        platform_scope = row.get("platform_scope")
+        if platform_scope is not None and (
+            not isinstance(platform_scope, list)
+            or not platform_scope
+            or any(not isinstance(item, str) or not item for item in platform_scope)
+        ):
+            _fail(
+                findings,
+                "v4_scorecard_dimension_platform_scope_invalid",
+                f"{rule_pack_id}/{key} platform_scope must be a non-empty string list",
+            )
+        score_source_path = row.get("score_source_path")
+        if score_source_path is not None and (not isinstance(score_source_path, str) or not score_source_path):
+            _fail(
+                findings,
+                "v4_scorecard_dimension_score_source_path_invalid",
+                f"{rule_pack_id}/{key} score_source_path must be non-empty string",
+            )
+        if not _positive_number(row.get("max_score")):
+            _fail(findings, "v4_scorecard_dimension_score_invalid", f"{rule_pack_id}/{key} max_score must be positive")
+        if not _positive_number(row.get("weight_percent")):
+            _fail(findings, "v4_scorecard_dimension_weight_invalid", f"{rule_pack_id}/{key} weight_percent must be positive")
+    duplicates = sorted({key for key in keys if key and keys.count(key) > 1})
+    if duplicates:
+        _fail(findings, "v4_scorecard_dimension_duplicate", f"{rule_pack_id} duplicate dimensions: {duplicates}")
+    active_keys = {row.get("key") for row in dimensions if isinstance(row, dict) and row.get("runtime_status") == "active"}
+    missing = sorted(V4_BASE_ACTIVE_DIMENSIONS - active_keys)
+    if missing:
+        _fail(findings, "v4_scorecard_dimensions_invalid", f"{rule_pack_id} active dimensions missing base keys: {missing}")
+
+
+def _check_v4_score_weight_profiles(
+    findings: list[dict[str, str]],
+    rule_pack_id: Any,
+    scorecard: dict[str, Any],
+) -> None:
+    profiles = scorecard.get("score_weight_profiles")
+    if not isinstance(profiles, dict):
+        _fail(findings, "v4_score_weight_profiles_invalid", f"{rule_pack_id} score_weight_profiles must be an object")
+        return
+    dimension_keys = {
+        row.get("key")
+        for row in scorecard.get("dimensions", [])
+        if isinstance(row, dict) and row.get("key")
+    }
+    if isinstance(profiles.get("profiles"), list):
+        _check_v4_generic_score_weight_profiles(findings, rule_pack_id, profiles["profiles"], dimension_keys)
+        return
+    missing = sorted(V4_REQUIRED_SCORE_WEIGHT_PROFILES - set(profiles))
+    if missing:
+        _fail(findings, "v4_score_weight_profiles_invalid", f"{rule_pack_id} missing score weight profiles: {missing}")
+    for profile_name, weights in profiles.items():
+        if not isinstance(weights, dict) or not weights:
+            _fail(findings, "v4_score_weight_profile_invalid", f"{rule_pack_id}/{profile_name} must be a non-empty object")
+            continue
+        for key, weight in weights.items():
+            if key not in dimension_keys:
+                _fail(
+                    findings,
+                    "v4_score_weight_profile_dimension_unknown",
+                    f"{rule_pack_id}/{profile_name} unknown dimension: {key}",
+                )
+            if not _positive_number(weight):
+                _fail(
+                    findings,
+                    "v4_score_weight_profile_weight_invalid",
+                    f"{rule_pack_id}/{profile_name}/{key} weight must be positive",
+                )
+
+
+def _check_v4_generic_score_weight_profiles(
+    findings: list[dict[str, str]],
+    rule_pack_id: Any,
+    profiles: list[Any],
+    dimension_keys: set[Any],
+) -> None:
+    aliases: set[str] = set()
+    priority_groups: dict[tuple[Any, Any, Any, int], list[str]] = {}
+    for profile in profiles:
+        if not isinstance(profile, dict):
+            _fail(findings, "v4_score_weight_profile_invalid", f"{rule_pack_id} profile row must be object")
+            continue
+        profile_id = profile.get("profile_id")
+        if not profile_id:
+            _fail(findings, "v4_score_weight_profile_invalid", f"{rule_pack_id} profile_id must be non-empty")
+            continue
+        aliases.update(profile.get("historical_alias_ids") or [])
+        if profile.get("approval_status") != "approved":
+            _fail(findings, "v4_score_weight_profile_approval_invalid", f"{rule_pack_id}/{profile_id} must be approved")
+        if profile.get("runtime_enabled") is not True:
+            _fail(findings, "v4_score_weight_profile_runtime_invalid", f"{rule_pack_id}/{profile_id} runtime_enabled must be true")
+        weights = profile.get("weights")
+        if not isinstance(weights, dict) or not weights:
+            _fail(findings, "v4_score_weight_profile_invalid", f"{rule_pack_id}/{profile_id} weights must be non-empty object")
+            continue
+        total = 0.0
+        for key, weight in weights.items():
+            if key not in dimension_keys:
+                _fail(
+                    findings,
+                    "v4_score_weight_profile_dimension_unknown",
+                    f"{rule_pack_id}/{profile_id} unknown dimension: {key}",
+                )
+            if not _positive_number(weight):
+                _fail(
+                    findings,
+                    "v4_score_weight_profile_weight_invalid",
+                    f"{rule_pack_id}/{profile_id}/{key} weight must be positive",
+                )
+            else:
+                total += float(weight)
+        normalization = profile.get("normalization_policy") or "sum_100"
+        if normalization == "raw_sum":
+            if not profile.get("normalization_reason"):
+                _fail(
+                    findings,
+                    "v4_score_weight_profile_normalization_invalid",
+                    f"{rule_pack_id}/{profile_id} raw_sum requires normalization_reason",
+                )
+        elif round(total, 6) != 100:
+            _fail(
+                findings,
+                "v4_score_weight_profile_total_invalid",
+                f"{rule_pack_id}/{profile_id} weights must sum to 100",
+            )
+        group = (
+            tuple(profile.get("platform_scope") or []),
+            profile.get("content_format") or "*",
+            profile.get("applies_when") or "always",
+            int(profile.get("priority") or 0),
+        )
+        priority_groups.setdefault(group, []).append(str(profile_id))
+    missing = sorted(V4_REQUIRED_SCORE_WEIGHT_PROFILES - aliases)
+    if missing:
+        _fail(findings, "v4_score_weight_profiles_invalid", f"{rule_pack_id} missing historical aliases: {missing}")
+    for group, ids in priority_groups.items():
+        if len(ids) > 1:
+            _fail(findings, "v4_score_weight_profile_priority_invalid", f"{rule_pack_id} priority conflict {group}: {ids}")
+
+
+def _check_v4_scorecard_rule_refs(
+    findings: list[dict[str, str]],
+    rule_pack_id: Any,
+    scorecard: dict[str, Any],
+) -> None:
+    dimension_keys = {
+        row.get("key")
+        for row in scorecard.get("dimensions", [])
+        if isinstance(row, dict) and row.get("key")
+    }
+    for rule in scorecard.get("scoring_rules", []):
+        if not isinstance(rule, dict):
+            _fail(findings, "v4_scoring_rule_invalid", f"{rule_pack_id} scoring rule must be object")
+            continue
+        if rule.get("dimension_key") not in dimension_keys:
+            _fail(
+                findings,
+                "v4_scoring_rule_dimension_unknown",
+                f"{rule_pack_id}/{rule.get('scoring_rule_id')} unknown dimension_key: {rule.get('dimension_key')}",
+            )
+
+
+def _positive_number(value: Any) -> bool:
+    return isinstance(value, (int, float)) and not isinstance(value, bool) and value > 0
+
+
 def _fail(findings: list[dict[str, str]], check_id: str, message: str) -> None:
     findings.append({"level": "fail", "check_id": check_id, "message": message})
 

scripts/validate_v5_m4_dead_config_archive.py

@@ -0,0 +1,76 @@
+"""Validate V5-M4 dead config archive manifests."""
+
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+from typing import Any
+
+
+ROOT = Path(__file__).resolve().parents[1]
+ARCHIVE_ROOT = ROOT / "archive/v5_m4_dead_config_archive"
+REQUIRED_FIELDS = {
+    "archive_id",
+    "source_path",
+    "archive_path",
+    "archived_at",
+    "archived_by",
+    "proof_no_runtime_read",
+    "restore_instruction",
+    "approval_status",
+}
+FORBIDDEN_SOURCE_PREFIXES = {
+    "sql/",
+    "tests/fixtures/cases/v4_douyin_57663",
+    "tests/fixtures/cases/v4_kuaishou_57758",
+}
+FORBIDDEN_SOURCE_PATHS = {
+    "tech_documents/数据接口与来源/score_thresholds.json",
+    "tech_documents/数据接口与来源/walk_policy.json",
+}
+
+
+def validate_archive(root: Path = ARCHIVE_ROOT) -> list[dict[str, Any]]:
+    findings: list[dict[str, Any]] = []
+    if not root.exists():
+        return findings
+    for manifest_path in sorted(root.glob("*/manifest.json")):
+        manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
+        for index, item in enumerate(manifest.get("items", [])):
+            _check_item(findings, manifest_path, index, item)
+    return findings
+
+
+def _check_item(findings: list[dict[str, Any]], manifest_path: Path, index: int, item: dict[str, Any]) -> None:
+    missing = sorted(field for field in REQUIRED_FIELDS if not item.get(field))
+    if missing:
+        _fail(findings, "archive_manifest_required_field", f"{manifest_path}:{index} missing {missing}")
+    source_path = str(item.get("source_path") or "")
+    archive_path = str(item.get("archive_path") or "")
+    source_file = source_path.split("#", 1)[0]
+    if source_file in FORBIDDEN_SOURCE_PATHS or any(source_file.startswith(prefix) for prefix in FORBIDDEN_SOURCE_PREFIXES):
+        _fail(findings, "archive_forbidden_source", f"{manifest_path}:{index} forbidden source_path {source_path}")
+    if archive_path:
+        path = ROOT / archive_path
+        if not path.exists():
+            _fail(findings, "archive_file_missing", f"{manifest_path}:{index} archive_path missing: {archive_path}")
+        if not path.is_relative_to(ARCHIVE_ROOT):
+            _fail(findings, "archive_path_outside_root", f"{manifest_path}:{index} archive_path outside M4 archive: {archive_path}")
+    if item.get("approval_status") != "approved":
+        _fail(findings, "archive_approval_status", f"{manifest_path}:{index} approval_status must be approved")
+
+
+def _fail(findings: list[dict[str, Any]], check_id: str, message: str) -> None:
+    findings.append({"level": "fail", "check_id": check_id, "message": message})
+
+
+def main() -> int:
+    findings = validate_archive()
+    result = {"status": "fail" if findings else "pass", "findings": findings}
+    print(json.dumps(result, ensure_ascii=False, indent=2))
+    return 1 if findings else 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/validate_walk_strategy_config.py

@@ -11,6 +11,7 @@ if str(ROOT) not in sys.path:
     sys.path.insert(0, str(ROOT))
 
 from content_agent.integrations.walk_strategy_json import (
+    EDGE_CATALOG_PATH,
     RULE_PACK_PATH,
     WALK_STRATEGY_PATH,
     validate_walk_strategy_config,
@@ -27,6 +28,7 @@ def main() -> int:
         root_dir=root_dir,
         strategy_path=path.relative_to(root_dir),
         rule_pack_path=RULE_PACK_PATH,
+        edge_catalog_path=args.edge_catalog_path,
     )
     result: dict[str, Any] = {
         "status": "fail" if any(finding["level"] == "fail" for finding in findings) else "pass",
@@ -46,6 +48,7 @@ def _parse_args() -> argparse.Namespace:
         type=Path,
         default=_repo_root() / WALK_STRATEGY_PATH,
     )
+    parser.add_argument("--edge-catalog-path", type=Path, default=EDGE_CATALOG_PATH)
     return parser.parse_args()