feat(M3): per-entity dispatch, config-driven portrait triage, scorecard missing dimensions

- M3A: 参数化 _select_dispatch(五元组),新增 CONFIG_RULE_PACK_DISPATCH_CONFLICT
  错误码与加载期冲突护栏;validator 增 dispatch_conflict 全局唯一性检查
- M3B: load_policy_bundle 暴露 rule_pack_by_entity(仅 enabled dispatch),
  顶层 Content shim 与 rule_judgment.run() 签名不变
- M3C: Excel→JSON 配置驱动画像止血——missing_content_portrait 改
  KEEP_CONTENT_FOR_REVIEW/review/pending,age_50_plus_weak 只匹配 ["weak"];
  新增 hard gate KEEP→pending effect mapping(双侧同步,byte-equal);
  修正 Excel meta strategy_id 漂移;evaluator 零业务硬编码分支
- M3D: scorecard dimension row 级 score_missing、missing_dimensions 复盘字段,
  单维度缺失 0 分走 threshold,全缺失才走 score_missing_policy
- M3E: 按 brief 逐字补 15 个测试 + portrait_reject 配置反证变体;
  real_id45 回放受控变化(4×KEEP/pending,回放口径:fake decode 成功,
  唯一拦截闸为画像缺失;DB 收割口径 1+3 已在 docstring 标注)

235 passed;config gate 5 闸全 pass;schema registry pass;无 DB/runtime 改动

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

content_agent/business_modules/rule_judgment/evaluator.py

@@ -57,7 +57,7 @@ def decide(
 
     scorecard = _scorecard_total(bundle, rule_pack.get("scorecard", {}))
     score = scorecard.get("total_score")
-    if score is None or not scorecard.get("matched_scoring_rules"):
+    if score is None or _scorecard_all_dimensions_missing(scorecard):
         return _missing_score_decision(
             run_id,
             policy_run_id,
@@ -152,6 +152,9 @@ def _build_decision(
             "strategy_version": policy_bundle["strategy_version"],
             "runtime_record_schema_version": RUNTIME_RECORD_SCHEMA_VERSION,
             "decision_evidence_refs": _evidence_refs(policy_bundle["rule_pack"]),
+            "missing_dimensions": [
+                row["key"] for row in scorecard.get("dimensions", []) if row.get("score_missing")
+            ],
             **replay_marker,
         },
     }
@@ -274,6 +277,7 @@ def _scorecard_total(bundle: dict[str, Any], config: dict[str, Any]) -> dict[str
                 "max_score": dimension.get("max_score"),
                 "score": score,
                 "matched_rule_id": rule.get("scoring_rule_id") if rule else None,
+                "score_missing": rule is None,
                 "evidence_paths": dimension.get("evidence_paths", []),
             }
         )
@@ -285,6 +289,14 @@ def _scorecard_total(bundle: dict[str, Any], config: dict[str, Any]) -> dict[str
     }
 
 
+def _scorecard_has_any_evidence(scorecard: dict[str, Any]) -> bool:
+    return any(not row.get("score_missing") for row in scorecard.get("dimensions", []))
+
+
+def _scorecard_all_dimensions_missing(scorecard: dict[str, Any]) -> bool:
+    return not _scorecard_has_any_evidence(scorecard)
+
+
 def _score_dimension(
     bundle: dict[str, Any], dimension: dict[str, Any], rules: list[dict[str, Any]]
 ) -> tuple[int | float, dict[str, Any] | None]:

content_agent/errors.py

@@ -18,6 +18,7 @@ class ErrorCode(StrEnum):
     PLATFORM_CONFIG_MISSING = "PLATFORM_CONFIG_MISSING"
     PLATFORM_REQUEST_FAILED = "PLATFORM_REQUEST_FAILED"
     QUERY_GENERATION_FAILED = "QUERY_GENERATION_FAILED"
+    CONFIG_RULE_PACK_DISPATCH_CONFLICT = "CONFIG_RULE_PACK_DISPATCH_CONFLICT"
 
 
 SENSITIVE_KEYS = {
@@ -111,5 +112,6 @@ def _safe_message(error_code: ErrorCode) -> str:
         ErrorCode.PLATFORM_CONFIG_MISSING: "platform config missing",
         ErrorCode.PLATFORM_REQUEST_FAILED: "platform request failed",
         ErrorCode.QUERY_GENERATION_FAILED: "query generation failed",
+        ErrorCode.CONFIG_RULE_PACK_DISPATCH_CONFLICT: "rule pack dispatch conflict in config",
     }
     return messages.get(error_code, error_code.value.lower())

content_agent/integrations/policy_json.py

@@ -4,6 +4,7 @@ from pathlib import Path
 from typing import Any
 
 from content_agent.constants import DEFAULT_POLICY_BUNDLE_ID, EVIDENCE_BUNDLE_SCHEMA_VERSION, RUNTIME_SCHEMA_VERSION
+from content_agent.errors import ContentAgentError, ErrorCode
 from content_agent.integrations import config_store
 from content_agent.integrations.walk_strategy_json import WalkStrategyStore
 
@@ -26,6 +27,7 @@ class JsonPolicyBundleStore:
 
         dispatch = _select_dispatch(rule_package, actual_strategy_version)
         rule_pack = _find_rule_pack_by_dispatch(rule_package, dispatch)
+        rule_pack_by_entity = _build_rule_pack_by_entity(rule_package, actual_strategy_version)
         walk_strategy = WalkStrategyStore(self.root_dir).load_walk_strategy()
         bundle = {
             "policy_bundle_id": DEFAULT_POLICY_BUNDLE_ID,
@@ -37,6 +39,7 @@ class JsonPolicyBundleStore:
             "rule_pack_version": rule_pack["version"],
             "dispatch": dispatch,
             "dispatch_id": dispatch["dispatch_id"],
+            "rule_pack_by_entity": rule_pack_by_entity,
             "runtime_stage": dispatch["runtime_stage"],
             "target_entity": dispatch["target_entity"],
             "content_format": dispatch["content_format"],
@@ -76,20 +79,82 @@ def _strategy_version(rule_package: dict[str, Any]) -> str:
     return binding_version or "V1"
 
 
-def _select_dispatch(rule_package: dict[str, Any], strategy_version: str) -> dict[str, Any]:
-    matches = [
+def _select_dispatch(
+    rule_package: dict[str, Any],
+    strategy_version: str,
+    *,
+    target_entity: str = "Content",
+    content_format: str = "video",
+    runtime_stage: str = "V1.0",
+    platform: str = "douyin",
+) -> dict[str, Any]:
+    matches = _enabled_dispatches(
+        rule_package,
+        strategy_version,
+        target_entity=target_entity,
+        content_format=content_format,
+        runtime_stage=runtime_stage,
+        platform=platform,
+    )
+    return _assert_single_enabled_dispatch(matches, target_entity=target_entity, content_format=content_format)
+
+
+def _enabled_dispatches(
+    rule_package: dict[str, Any],
+    strategy_version: str,
+    *,
+    target_entity: str,
+    content_format: str,
+    runtime_stage: str,
+    platform: str,
+) -> list[dict[str, Any]]:
+    return [
         dispatch
         for dispatch in rule_package.get("rule_pack_dispatch", [])
         if dispatch.get("dispatch_enabled")
-        and dispatch.get("platform") == "douyin"
-        and dispatch.get("runtime_stage") == "V1.0"
+        and dispatch.get("platform") == platform
+        and dispatch.get("runtime_stage") == runtime_stage
         and dispatch.get("strategy_version") == strategy_version
-        and dispatch.get("target_entity") == "Content"
-        and dispatch.get("content_format") == "video"
+        and dispatch.get("target_entity") == target_entity
+        and dispatch.get("content_format") == content_format
     ]
-    if len(matches) != 1:
-        raise ValueError(f"expected exactly one Content dispatch, got {len(matches)}")
-    return matches[0]
+
+
+def _assert_single_enabled_dispatch(
+    matches: list[dict[str, Any]],
+    *,
+    target_entity: str,
+    content_format: str,
+) -> dict[str, Any]:
+    if len(matches) == 1:
+        return matches[0]
+    if not matches:
+        raise ValueError(f"dispatch not found for {target_entity}/{content_format}")
+    conflict_rule_pack_ids = [dispatch.get("rule_pack_id") for dispatch in matches]
+    raise ContentAgentError(
+        ErrorCode.CONFIG_RULE_PACK_DISPATCH_CONFLICT,
+        f"CONFIG_RULE_PACK_DISPATCH_CONFLICT: multiple enabled dispatches for "
+        f"{target_entity}/{content_format}: {conflict_rule_pack_ids}",
+        {"target_entity": target_entity, "content_format": content_format,
+         "conflict_rule_pack_ids": conflict_rule_pack_ids},
+    )
+
+
+def _build_rule_pack_by_entity(rule_package: dict[str, Any], strategy_version: str) -> dict[str, Any]:
+    by_entity: dict[str, Any] = {}
+    for dispatch in rule_package.get("rule_pack_dispatch", []):
+        if not (
+            dispatch.get("dispatch_enabled")
+            and dispatch.get("platform") == "douyin"
+            and dispatch.get("runtime_stage") == "V1.0"
+            and dispatch.get("strategy_version") == strategy_version
+        ):
+            continue
+        by_entity[dispatch["target_entity"]] = {
+            "dispatch": dispatch,
+            "rule_pack": _find_rule_pack_by_dispatch(rule_package, dispatch),
+        }
+    return by_entity
 
 
 def _find_rule_pack_by_dispatch(rule_package: dict[str, Any], dispatch: dict[str, Any]) -> dict[str, Any]:

product_documents/规则包/douyin_rule_packs.v1.json

@@ -390,9 +390,9 @@
             "field": "content_audience_profile",
             "op": "is_empty"
           },
-          "decision_action": "REJECT_CONTENT",
+          "decision_action": "KEEP_CONTENT_FOR_REVIEW",
           "decision_reason_code": "missing_content_portrait",
-          "severity": "fatal",
+          "severity": "review",
           "stop_scoring": true,
           "priority": 50
         },
@@ -403,8 +403,7 @@
             "field": "content_audience_profile.age_50_plus_level",
             "op": "in",
             "value": [
-              "weak",
-              "missing"
+              "weak"
             ]
           },
           "decision_action": "REJECT_CONTENT",
@@ -1297,6 +1296,18 @@
       "priority": 20,
       "enabled": true
     },
+    {
+      "mapping_id": "map_keep_for_review_pending_hard_gate",
+      "target_level": "content",
+      "decision_action": "KEEP_CONTENT_FOR_REVIEW",
+      "reason_category": "hard_gate",
+      "is_hard_gate": true,
+      "content_effect_status": "pending",
+      "query_effect_status": "pending",
+      "priority": 25,
+      "enabled": true,
+      "notes": "画像缺失等 hard gate 直接给待复看时，内容与 query 状态都记 pending。"
+    },
     {
       "mapping_id": "map_reject_rule_blocked",
       "target_level": "content",

scripts/validate_rule_pack_config.py

@@ -35,10 +35,19 @@ def validate_rule_pack_config(pkg: dict[str, Any]) -> list[dict[str, Any]]:
         actions.update(entry.get("allowed_actions", []))
     reason_codes = {r.get("decision_reason_code") for r in pkg.get("decision_reason_codes", [])}
 
+    enabled_by_group: dict[tuple[Any, ...], list[str]] = {}
     for dispatch in pkg.get("rule_pack_dispatch", []):
         if dispatch.get("rule_pack_id") not in rule_pack_ids:
             _fail(findings, "dispatch_rule_pack_id",
                   f"{dispatch.get('dispatch_id')} references unknown rule_pack_id: {dispatch.get('rule_pack_id')}")
+        if dispatch.get("dispatch_enabled"):
+            group = (dispatch.get("platform"), dispatch.get("strategy_version"), dispatch.get("runtime_stage"),
+                     dispatch.get("target_entity"), dispatch.get("content_format"))
+            enabled_by_group.setdefault(group, []).append(dispatch.get("rule_pack_id"))
+    for group, pack_ids in enabled_by_group.items():
+        if len(pack_ids) > 1:
+            _fail(findings, "dispatch_conflict",
+                  f"CONFIG_RULE_PACK_DISPATCH_CONFLICT: multiple enabled dispatches for group {group}: {pack_ids}")
 
     for pack in rule_packs:
         pid = pack.get("rule_pack_id")

tests/fixtures/snapshots/matrix/real_id45__default.json

@@ -1,8 +1,8 @@
 {
   "effect_status_counts": {
     "failed": 0,
-    "pending": 0,
-    "rule_blocked": 4,
+    "pending": 4,
+    "rule_blocked": 0,
     "success": 0
   },
   "pooled": 0,
@@ -12,5 +12,5 @@
     "missing_content_portrait",
     "missing_content_portrait"
   ],
-  "rejected": 4
+  "rejected": 0
 }

tests/fixtures/snapshots/matrix/real_id45__portrait_reject.json

@@ -0,0 +1,16 @@
+{
+  "effect_status_counts": {
+    "failed": 0,
+    "pending": 0,
+    "rule_blocked": 4,
+    "success": 0
+  },
+  "pooled": 0,
+  "reasons": [
+    "missing_content_portrait",
+    "missing_content_portrait",
+    "missing_content_portrait",
+    "missing_content_portrait"
+  ],
+  "rejected": 4
+}

tests/fixtures/snapshots/real_id45/decision_summary.json

@@ -1,6 +1,6 @@
 {
   "pending_content_count": 0,
   "pooled_content_count": 0,
-  "rejected_content_count": 4,
-  "review_content_count": 0
+  "rejected_content_count": 0,
+  "review_content_count": 4
 }

tests/test_case_replay.py

@@ -1,7 +1,10 @@
 """Real + synthetic case replay tests (V2-M0D).
 
-- real_id45: the harvested production baseline (demand_content.id=45) replays to
-  the captured all-REJECT outcome. This is the regression anchor M2/M3 will move.
+- real_id45: the harvested production baseline (demand_content.id=45). Pre-M3 it
+  replayed to all-REJECT; M3C (portrait missing -> KEEP_CONTENT_FOR_REVIEW/pending)
+  moved it to all-KEEP, because replay recomputes pattern recall with fake success
+  clients so the portrait gate is the only blocker (harvested DB facts differ:
+  1 missing_content_portrait + 3 content_pattern_recall_required).
 - syn_pool / syn_review: synthetic corpora (authored with full portrait fields)
   exercise the ADD / KEEP paths the real baseline cannot (its portrait is empty).
 
@@ -71,13 +74,14 @@ def _synthetic_item(content_id: str, *, age_level: str, digg: int) -> dict[str,
     }
 
 
-def test_replay_id45_baseline_all_reject(tmp_path):
+def test_replay_id45_baseline_portrait_review(tmp_path):
     artifacts = replay_case("real_id45", runtime_root=tmp_path / "rt")
     assert artifacts.state["status"] == "success"
-    # Reproduces the captured real outcome: every content rejected, none pooled.
-    assert artifacts.summary["rejected_content_count"] == 4
+    # M3C 受控变化: portrait-missing content goes to review instead of reject.
+    assert artifacts.summary["review_content_count"] == 4
+    assert artifacts.summary["rejected_content_count"] == 0
     assert artifacts.summary["pooled_content_count"] == 0
-    assert _decision_counts(artifacts) == {"REJECT_CONTENT": 4}
+    assert _decision_counts(artifacts) == {"KEEP_CONTENT_FOR_REVIEW": 4}
     assert_matches("real_id45/decision_summary", artifacts.summary, subset_keys=_SUMMARY_KEYS)
 
 

tests/test_config_case_matrix.py

@@ -47,15 +47,32 @@ def _outcome(artifacts) -> dict:
     }
 
 
+def _portrait_reject_store(root: Path) -> JsonPolicyBundleStore:
+    """M3C counterproof variant: flip missing_content_portrait back to REJECT by config only."""
+    (root / _RULE_PACK_REL).parent.mkdir(parents=True, exist_ok=True)
+    (root / _WALK_REL).parent.mkdir(parents=True, exist_ok=True)
+    shutil.copy(ROOT / _WALK_REL, root / _WALK_REL)
+    package = json.loads((ROOT / _RULE_PACK_REL).read_text(encoding="utf-8"))
+    for pack in package.get("rule_packs", []):
+        for gate in pack.get("hard_gates", []):
+            if gate.get("gate_id") == "missing_content_portrait":
+                gate["decision_action"] = "REJECT_CONTENT"
+                gate["severity"] = "fatal"
+    (root / _RULE_PACK_REL).write_text(json.dumps(package, ensure_ascii=False, indent=2), encoding="utf-8")
+    return JsonPolicyBundleStore(root)
+
+
 def _variant_overrides(variant: str, cfg_dir: Path):
     if variant == "default":
         return None
     if variant == "relaxed_portrait":
         return {"policy_store": _relaxed_portrait_store(cfg_dir)}
+    if variant == "portrait_reject":
+        return {"policy_store": _portrait_reject_store(cfg_dir)}
     raise ValueError(variant)
 
 
-@pytest.mark.parametrize("variant", ["default", "relaxed_portrait"])
+@pytest.mark.parametrize("variant", ["default", "relaxed_portrait", "portrait_reject"])
 def test_matrix_real_id45(variant, tmp_path):
     overrides = _variant_overrides(variant, tmp_path / "cfg")
     artifacts = replay_case("real_id45", runtime_root=tmp_path / "rt", config_overrides=overrides)
@@ -76,8 +93,10 @@ def test_relaxed_portrait_changes_outcome(tmp_path):
     assert base != relaxed
     assert "missing_content_portrait" in base["reasons"]
     assert "missing_content_portrait" not in relaxed["reasons"]
-    assert base["effect_status_counts"]["rule_blocked"] == 4
-    assert relaxed["effect_status_counts"]["rule_blocked"] == 0
+    # M3C 受控变化: default config now parks portrait-missing content as pending.
+    assert base["effect_status_counts"]["pending"] == 4
+    assert base["effect_status_counts"]["rule_blocked"] == 0
+    assert relaxed["effect_status_counts"]["failed"] == 4
 
 
 def test_matrix_query_profile_variant():
@@ -87,9 +106,8 @@ def test_matrix_query_profile_variant():
     assert validate_query_prompts_config(config) == []
 
 
-@pytest.mark.xfail(reason="V2-M3: per-entity dispatch (Content hardcode) not removed yet", strict=True)
 def test_decoupling_counterproof():
-    # When M3 parametrizes dispatch by target_entity, the Content hardcode is gone
-    # and a non-Content (e.g. Author) pack can be routed without falling back.
+    # M3A removed the Content hardcode: dispatch is parametrized by target_entity,
+    # so a non-Content (e.g. Author) pack can be routed without falling back.
     source = (ROOT / "content_agent/integrations/policy_json.py").read_text(encoding="utf-8")
     assert 'target_entity") == "Content"' not in source

tests/test_config_tooling.py

@@ -38,6 +38,41 @@ def test_rule_pack_fk_validator_has_no_failures():
     assert [f for f in findings if f["level"] == "fail"] == []
 
 
+def test_rule_pack_validator_rejects_enabled_dispatch_conflict():
+    import json
+    from copy import deepcopy
+
+    mod = _load_script("validate_rule_pack_config")
+    pkg = deepcopy(json.loads(RULE_PACK_JSON.read_text("utf-8")))
+    duplicate = dict(pkg["rule_pack_dispatch"][0])
+    duplicate["dispatch_id"] = "dispatch_content_duplicate"
+    pkg["rule_pack_dispatch"].append(duplicate)
+
+    findings = mod.validate_rule_pack_config(pkg)
+
+    conflicts = [f for f in findings if f["check_id"] == "dispatch_conflict"]
+    assert len(conflicts) == 1
+    assert conflicts[0]["level"] == "fail"
+    assert "CONFIG_RULE_PACK_DISPATCH_CONFLICT" in conflicts[0]["message"]
+    assert "douyin_content_discovery_rule_pack_v1" in conflicts[0]["message"]
+
+
+def test_excel_meta_strategy_id_matches_walk_strategy():
+    import json
+
+    openpyxl = pytest.importorskip("openpyxl")
+    workbook = openpyxl.load_workbook(
+        ROOT / "tech_documents/规则包映射/规则包映射配置表.xlsx", read_only=True
+    )
+    rows = list(workbook["rule_package_meta"].iter_rows(values_only=True))
+    meta = dict(zip(rows[0], rows[2]))  # row 2 is the data-dictionary row; row 3 is data
+    walk_strategy = json.loads(
+        (ROOT / "product_documents/抖音游走策略/douyin_walk_strategy.v1.json").read_text("utf-8")
+    )
+
+    assert meta["strategy_id"] == walk_strategy["strategy_id"]
+
+
 def test_query_prompts_validator_passes_after_m2():
     mod = _load_script("validate_query_prompts_config")
     assert mod.main() == 0

tests/test_policy_dispatch.py

@@ -2,11 +2,15 @@ from __future__ import annotations
 
 import json
 import shutil
+from copy import deepcopy
 from pathlib import Path
 
 import pytest
 
-from content_agent.integrations.policy_json import JsonPolicyBundleStore
+from content_agent.errors import ContentAgentError, ErrorCode
+from content_agent.integrations.policy_json import JsonPolicyBundleStore, _select_dispatch
+
+RULE_PACK_JSON = Path("product_documents/规则包/douyin_rule_packs.v1.json")
 
 
 def test_policy_bundle_uses_content_dispatch_and_exports_runtime_contracts():
@@ -37,11 +41,11 @@ def test_policy_bundle_fails_when_content_dispatch_is_missing(tmp_path):
             dispatch["dispatch_enabled"] = False
     path.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")
 
-    with pytest.raises(ValueError, match="expected exactly one Content dispatch"):
+    with pytest.raises(ValueError, match="dispatch not found for Content/video"):
         JsonPolicyBundleStore(root).load_policy_bundle("V1")
 
 
-def test_policy_bundle_fails_when_content_dispatch_is_ambiguous(tmp_path):
+def test_dispatch_conflict_raises_config_error_with_rule_pack_ids(tmp_path):
     root = _copy_policy_files(tmp_path)
     path = root / "product_documents/规则包/douyin_rule_packs.v1.json"
     data = json.loads(path.read_text(encoding="utf-8"))
@@ -51,8 +55,74 @@ def test_policy_bundle_fails_when_content_dispatch_is_ambiguous(tmp_path):
     data["rule_pack_dispatch"].append(duplicate)
     path.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")
 
-    with pytest.raises(ValueError, match="expected exactly one Content dispatch"):
+    with pytest.raises(ContentAgentError) as exc_info:
         JsonPolicyBundleStore(root).load_policy_bundle("V1")
+    error = exc_info.value
+    assert error.error_code == ErrorCode.CONFIG_RULE_PACK_DISPATCH_CONFLICT
+    assert "douyin_content_discovery_rule_pack_v1" in error.message
+    assert error.detail["conflict_rule_pack_ids"] == [
+        "douyin_content_discovery_rule_pack_v1",
+        "douyin_content_discovery_rule_pack_v1",
+    ]
+
+
+def test_select_dispatch_still_returns_content_for_default_bundle():
+    rule_package = json.loads(RULE_PACK_JSON.read_text(encoding="utf-8"))
+
+    dispatch = _select_dispatch(rule_package, "V1")
+
+    assert dispatch["dispatch_id"] == "dispatch_content"
+    assert dispatch["target_entity"] == "Content"
+
+
+def test_select_dispatch_can_select_non_content_when_enabled():
+    rule_package = json.loads(RULE_PACK_JSON.read_text(encoding="utf-8"))
+    rule_package = deepcopy(rule_package)
+    author = next(d for d in rule_package["rule_pack_dispatch"] if d["target_entity"] == "Author")
+    author["dispatch_enabled"] = True
+    author["runtime_stage"] = "V1.0"
+    author["strategy_version"] = "V1"
+
+    dispatch = _select_dispatch(
+        rule_package, "V1", target_entity="Author", content_format=author["content_format"]
+    )
+
+    assert dispatch["rule_pack_id"] == "douyin_author_expand_rule_pack_v1"
+
+
+def test_load_policy_bundle_keeps_content_shim():
+    bundle = JsonPolicyBundleStore().load_policy_bundle("V1")
+
+    assert bundle["target_entity"] == "Content"
+    assert bundle["rule_pack_id"] == "douyin_content_discovery_rule_pack_v1"
+    assert bundle["rule_pack"] is bundle["rule_pack_by_entity"]["Content"]["rule_pack"]
+
+
+def test_load_policy_bundle_exposes_rule_pack_by_entity():
+    bundle = JsonPolicyBundleStore().load_policy_bundle("V1")
+
+    by_entity = bundle["rule_pack_by_entity"]
+    assert set(by_entity) == {"Content"}
+    assert by_entity["Content"]["dispatch"]["dispatch_id"] == "dispatch_content"
+    assert by_entity["Content"]["rule_pack"]["rule_pack_id"] == bundle["rule_pack_id"]
+
+
+def test_enabled_author_dispatch_can_be_found_by_entity_without_replacing_content_shim(tmp_path):
+    root = _copy_policy_files(tmp_path)
+    path = root / "product_documents/规则包/douyin_rule_packs.v1.json"
+    data = json.loads(path.read_text(encoding="utf-8"))
+    author = next(d for d in data["rule_pack_dispatch"] if d["target_entity"] == "Author")
+    author["dispatch_enabled"] = True
+    author["runtime_stage"] = "V1.0"
+    author["strategy_version"] = "V1"
+    path.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")
+
+    bundle = JsonPolicyBundleStore(root).load_policy_bundle("V1")
+
+    assert bundle["rule_pack_id"] == "douyin_content_discovery_rule_pack_v1"
+    assert set(bundle["rule_pack_by_entity"]) == {"Content", "Author"}
+    author_pack = bundle["rule_pack_by_entity"]["Author"]["rule_pack"]
+    assert author_pack["rule_pack_id"] == "douyin_author_expand_rule_pack_v1"
 
 
 def test_policy_bundle_fails_when_dispatch_points_to_missing_rule_pack(tmp_path):
@@ -88,4 +158,9 @@ def _copy_policy_files(tmp_path: Path) -> Path:
         "product_documents/规则包/douyin_rule_packs.v1.json",
         root / "product_documents/规则包/douyin_rule_packs.v1.json",
     )
+    (root / "product_documents/抖音游走策略").mkdir(parents=True)
+    shutil.copy(
+        "product_documents/抖音游走策略/douyin_walk_strategy.v1.json",
+        root / "product_documents/抖音游走策略/douyin_walk_strategy.v1.json",
+    )
     return root

tests/test_query_effect_aggregation.py

@@ -22,7 +22,8 @@ def test_search_clues_aggregate_query_effect_status_from_decisions(tmp_path):
     }
 
     assert clues["q_001"]["search_query_effect_status"] == "success"
-    assert clues["q_001"]["effect_status_counts"] == {"success": 1, "rule_blocked": 1}
+    # M3C 受控变化: the portrait-missing mock content is pending (was rule_blocked).
+    assert clues["q_001"]["effect_status_counts"] == {"success": 1, "pending": 1}
     assert clues["q_001"]["query_aggregation_id"] == "agg_query_success"
     assert clues["q_001"]["raw_payload"]["query_aggregation_id"] == "agg_query_success"
     assert clues["q_001"]["walk_next_step"] == "keep_or_consider_next_page"

tests/test_rule_judgment_hard_gates.py

@@ -57,6 +57,62 @@ def test_pattern_recall_pending_cannot_pass_hard_gate(tmp_path):
     assert decision["search_query_effect_status"] == "rule_blocked"
 
 
+def test_hard_gate_action_is_taken_from_rule_config(tmp_path):
+    state = _state(tmp_path)
+    bundle = deepcopy(state["evidence_bundles"][0])
+    bundle["content_audience_profile"] = {}
+
+    decision = decide(state["run_id"], state["policy_run_id"], 1, bundle, state["policy_bundle"])
+    assert decision["decision_action"] == "KEEP_CONTENT_FOR_REVIEW"
+
+    flipped = deepcopy(state["policy_bundle"])
+    for gate in flipped["rule_pack"]["hard_gates"]:
+        if gate["gate_id"] == "missing_content_portrait":
+            gate["decision_action"] = "REJECT_CONTENT"
+            gate["severity"] = "fatal"
+    decision = decide(state["run_id"], state["policy_run_id"], 1, bundle, flipped)
+    assert decision["decision_action"] == "REJECT_CONTENT"
+    assert decision["search_query_effect_status"] == "rule_blocked"
+
+
+def test_missing_portrait_review_is_config_driven_not_code_special_case(tmp_path):
+    state = _state(tmp_path)
+    bundle = deepcopy(state["evidence_bundles"][0])
+    bundle["content_audience_profile"] = {}
+
+    decision = decide(state["run_id"], state["policy_run_id"], 1, bundle, state["policy_bundle"])
+
+    assert decision["decision_action"] == "KEEP_CONTENT_FOR_REVIEW"
+    assert decision["decision_reason_code"] == "missing_content_portrait"
+    assert decision["search_query_effect_status"] == "pending"
+    assert decision["triggered_blocking_rules"] == ["missing_content_portrait"]
+    assert decision["score"] is None
+    assert decision["decision_replay_data"]["effect_mapping_id"] == "map_keep_for_review_pending_hard_gate"
+
+
+def test_age_50_plus_weak_still_rejects_weak_only(tmp_path):
+    state = _state(tmp_path)
+    bundle = deepcopy(state["evidence_bundles"][0])
+    bundle["content_audience_profile"]["age_50_plus_level"] = "weak"
+
+    decision = decide(state["run_id"], state["policy_run_id"], 1, bundle, state["policy_bundle"])
+
+    assert decision["decision_action"] == "REJECT_CONTENT"
+    assert decision["decision_reason_code"] == "age_50_plus_weak"
+    assert decision["search_query_effect_status"] == "rule_blocked"
+
+
+def test_missing_value_no_longer_hits_age_50_plus_weak_gate(tmp_path):
+    state = _state(tmp_path)
+    bundle = deepcopy(state["evidence_bundles"][0])
+    bundle["content_audience_profile"]["age_50_plus_level"] = "missing"
+
+    decision = decide(state["run_id"], state["policy_run_id"], 1, bundle, state["policy_bundle"])
+
+    assert "age_50_plus_weak" not in decision["triggered_blocking_rules"]
+    assert decision["decision_reason_code"] != "age_50_plus_weak"
+
+
 def _state(tmp_path):
     service = RunService(
         runtime_root=tmp_path / "runtime" / "v1",

tests/test_rule_judgment_scorecard.py

@@ -103,6 +103,52 @@ def test_scoring_rule_unknown_operator_fails_fast(tmp_path):
         )
 
 
+def test_single_missing_dimension_scores_zero_and_keeps_threshold_flow(tmp_path):
+    state = _state(tmp_path)
+    bundle = deepcopy(state["evidence_bundles"][0])
+    bundle["content_engagement_metrics"]["statistics"] = {}
+
+    decision = decide(state["run_id"], state["policy_run_id"], 1, bundle, state["policy_bundle"])
+
+    dimensions = {row["key"]: row for row in decision["scorecard"]["dimensions"]}
+    assert dimensions["interaction_performance"]["score_missing"] is True
+    assert dimensions["interaction_performance"]["score"] == 0
+    assert dimensions["content_audience_profile"]["score_missing"] is False
+    assert decision["score"] == 47
+    assert decision["decision_reason_code"] != "missing_score"
+    assert decision["scorecard"]["score_missing"] is False
+
+
+def test_all_dimensions_missing_uses_score_missing_policy(tmp_path):
+    state = _state(tmp_path)
+    bundle = deepcopy(state["evidence_bundles"][0])
+    bundle["content_audience_profile"]["age_50_plus_level"] = "unknown"
+    bundle["content_engagement_metrics"]["statistics"] = {}
+    bundle["content_risk_check"]["availability"] = "metadata_only"
+
+    decision = decide(state["run_id"], state["policy_run_id"], 1, bundle, state["policy_bundle"])
+
+    assert decision["decision_action"] == "REJECT_CONTENT"
+    assert decision["decision_reason_code"] == "missing_score"
+    assert decision["score"] is None
+    assert decision["scorecard"]["score_missing"] is True
+    assert all(row["score_missing"] for row in decision["scorecard"]["dimensions"])
+
+
+def test_dimension_missing_metadata_is_recorded(tmp_path):
+    state = _state(tmp_path)
+    bundle = deepcopy(state["evidence_bundles"][0])
+    bundle["content_engagement_metrics"]["statistics"] = {}
+
+    decision = decide(state["run_id"], state["policy_run_id"], 1, bundle, state["policy_bundle"])
+    assert decision["decision_replay_data"]["missing_dimensions"] == ["interaction_performance"]
+
+    full_decision = decide(
+        state["run_id"], state["policy_run_id"], 2, state["evidence_bundles"][0], state["policy_bundle"]
+    )
+    assert full_decision["decision_replay_data"]["missing_dimensions"] == []
+
+
 def _state(tmp_path):
     service = RunService(
         runtime_root=tmp_path / "runtime" / "v1",

tests/test_rule_pack_reading.py

@@ -102,3 +102,13 @@ def test_rule_judgment_does_not_call_pattern_recall_integrations():
     assert "decode_api" not in text
     assert "category_match" not in text
     assert "match-paths" not in text
+
+
+def test_disabled_future_packs_not_marked_as_active_entity_packs():
+    from content_agent.integrations.policy_json import JsonPolicyBundleStore
+
+    bundle = JsonPolicyBundleStore().load_policy_bundle("V1")
+
+    assert set(bundle["rule_pack_by_entity"]) == {"Content"}
+    for entity in ("Author", "Hashtag", "Path", "Budget"):
+        assert entity not in bundle["rule_pack_by_entity"]

tests/test_v1_graph.py

@@ -23,14 +23,15 @@ def test_v1_graph_generates_all_runtime_files(tmp_path):
     final_output = service.read_json(run_id, "final_output.json")
     assert final_output["policy_run_id"] == state["policy_run_id"]
     assert final_output["summary"]["pooled_content_count"] == 1
-    assert final_output["summary"]["review_content_count"] == 1
+    # M3C 受控变化: the portrait-missing mock content moved to review (was rejected).
+    assert final_output["summary"]["review_content_count"] == 2
     assert final_output["summary"]["pending_content_count"] == 0
-    assert final_output["summary"]["rejected_content_count"] == 1
+    assert final_output["summary"]["rejected_content_count"] == 0
     assert final_output["summary"]["effect_status_counts"] == {
         "success": 1,
-        "pending": 1,
+        "pending": 2,
         "failed": 0,
-        "rule_blocked": 1,
+        "rule_blocked": 0,
     }
     assert (
         final_output["policy"]["policy_bundle_hash"]