Align V4 validator with score thresholds

content_agent/business_modules/rule_judgment/evaluator.py

@@ -195,7 +195,13 @@ def _decide_v4(
             "matched_gates": [_gate_replay(gate) for gate in matched_gates],
             "effect_mapping_id": effect.get("mapping_id"),
             "reason_category": reason_category,
-            **_v4_walk_gate(allow_walk, decision_reason_code, scorecard=scorecard, score=score),
+            **_v4_walk_gate(
+                allow_walk,
+                decision_reason_code,
+                scorecard=scorecard,
+                score=score,
+                thresholds=thresholds,
+            ),
         },
     )
 
@@ -334,8 +340,10 @@ def _v4_walk_gate(
     *,
     scorecard: dict[str, Any] | None = None,
     score: Any = None,
+    thresholds: dict[str, Any] | None = None,
 ) -> dict[str, Any]:
     scorecard = scorecard or {}
+    thresholds = thresholds or _V4_THRESHOLDS_FALLBACK
     snapshot = {
         "query_relevance_score": scorecard.get("query_relevance_score"),
         "platform_performance_score": scorecard.get("platform_performance_score"),
@@ -343,7 +351,11 @@ def _v4_walk_gate(
     }
     return {
         "allow_walk": allow_walk,
-        "allow_walk_reason": "query>=70/platform>=65/score>=70" if allow_walk else reason,
+        "allow_walk_reason": (
+            f"query>={thresholds['walk_query']}/platform>={thresholds['walk_platform']}/score>={thresholds['walk_total']}"
+            if allow_walk
+            else reason
+        ),
         "walk_gate_snapshot": snapshot,
     }
 

content_agent/business_modules/run_record/validation.py

@@ -67,6 +67,15 @@ EFFECT_STATUSES = {"success", "pending", "failed", "rule_blocked"}
 WALK_STATUSES = {"success", "pending", "failed", "skipped", "rule_blocked"}
 V4_SCORECARD_SCHEMA_VERSION = "v4_scorecard.v1"
 V4_GEMINI_QUERY_RELEVANCE_SCHEMA_VERSION = "v4_gemini_query_relevance.v1"
+V4_SCORE_THRESHOLD_FALLBACK = {
+    "pool_total": 70,
+    "pool_query": 70,
+    "review_total": 55,
+    "review_query": 55,
+    "walk_query": 70,
+    "walk_platform": 65,
+    "walk_total": 70,
+}
 V4_LEGACY_FIELD_BLOCKLIST = {
     "fit_senior_50plus",
     "fit_confidence",
@@ -898,6 +907,22 @@ def _v4_score_values(decision: dict[str, Any]) -> tuple[Any, Any, Any]:
     )
 
 
+def _v4_score_thresholds(decision: dict[str, Any]) -> dict[str, Any]:
+    scorecard = decision.get("scorecard") or {}
+    thresholds = scorecard.get("score_thresholds")
+    if not isinstance(thresholds, dict):
+        return dict(V4_SCORE_THRESHOLD_FALLBACK)
+    return {**V4_SCORE_THRESHOLD_FALLBACK, **thresholds}
+
+
+def _v4_walk_threshold_message(thresholds: dict[str, Any]) -> str:
+    return (
+        f"query>={thresholds['walk_query']}/"
+        f"platform>={thresholds['walk_platform']}/"
+        f"score>={thresholds['walk_total']}"
+    )
+
+
 def _check_v4_score_contract(data: dict[str, Any], findings: list[dict[str, Any]]) -> None:
     for decision in data.get("rule_decisions.jsonl", []):
         if not _is_v4_contract_record(decision):
@@ -974,19 +999,20 @@ def _check_v4_walk_gate_contract(data: dict[str, Any], findings: list[dict[str,
             )
             continue
         query_score, platform_score, total_score = _v4_score_values(decision)
+        thresholds = _v4_score_thresholds(decision)
         expected_allow_walk = (
             _is_number(query_score)
             and _is_number(platform_score)
             and _is_number(total_score)
-            and query_score >= 70
-            and platform_score >= 65
-            and total_score >= 70
+            and query_score >= thresholds["walk_query"]
+            and platform_score >= thresholds["walk_platform"]
+            and total_score >= thresholds["walk_total"]
         )
         if replay_data["allow_walk"] != expected_allow_walk:
             _fail(
                 findings,
                 "v4_allow_walk_threshold_mismatch",
-                f"decision {decision.get('decision_id')} allow_walk does not match query>=70/platform>=65/score>=70",
+                f"decision {decision.get('decision_id')} allow_walk does not match {_v4_walk_threshold_message(thresholds)}",
             )
 
 
@@ -1218,14 +1244,20 @@ def _check_v4_action_thresholds(data: dict[str, Any], findings: list[dict[str, A
         if not (_is_number(query_score) and _is_number(total_score)):
             continue
         action = decision.get("decision_action")
+        thresholds = _v4_score_thresholds(decision)
         if action == "ADD_TO_CONTENT_POOL":
-            ok = query_score >= 70 and total_score >= 70
+            ok = query_score >= thresholds["pool_query"] and total_score >= thresholds["pool_total"]
         elif action == "KEEP_CONTENT_FOR_REVIEW":
-            ok = query_score >= 55 and total_score >= 55 and not (
-                query_score >= 70 and total_score >= 70
+            ok = (
+                query_score >= thresholds["review_query"]
+                and total_score >= thresholds["review_total"]
+                and not (
+                    query_score >= thresholds["pool_query"]
+                    and total_score >= thresholds["pool_total"]
+                )
             )
         elif action == "REJECT_CONTENT":
-            ok = query_score < 55 or total_score < 55
+            ok = query_score < thresholds["review_query"] or total_score < thresholds["review_total"]
         else:
             continue
         if not ok:

content_agent/flow_ledger_service.py

@@ -92,6 +92,12 @@ _FLOW_LEDGER_CACHE: dict[tuple[str, str], tuple[float, dict[str, Any]]] = {}
 _DEFAULT_CACHE_TTL_SECONDS = 300
 
 
+def _reason_label(reason: str, default: str) -> str:
+    if reason.startswith("query>=") and "/platform>=" in reason and "/score>=" in reason:
+        return "评分达到继续扩展门槛"
+    return REASON_LABELS.get(reason, default)
+
+
 class FlowLedgerService:
     def __init__(self, runtime: RuntimeStore) -> None:
         dashboard = DashboardService.from_runtime(runtime)
@@ -549,7 +555,7 @@ class FlowLedgerService:
             "action": action,
             "label": ACTION_LABELS.get(action, "待判断"),
             "reason": reason,
-            "reason_label": REASON_LABELS.get(reason, "系统原因待补充"),
+            "reason_label": _reason_label(reason, "系统原因待补充"),
             "score": score,
             "score_label": f"{round(float(score))} 分" if isinstance(score, (int, float)) else "",
             "score_items": _decision_score_items(scorecard, score),
@@ -579,7 +585,7 @@ class FlowLedgerService:
             "technical_retry_count": actions["TECHNICAL_RETRY_REQUIRED"],
             "reject_count": actions["REJECT_CONTENT"],
             "primary_reason": primary_reason,
-            "primary_reason_label": REASON_LABELS.get(primary_reason, "系统原因待补充"),
+            "primary_reason_label": _reason_label(primary_reason, "系统原因待补充"),
             "score_summary": _score_summary(decisions),
             "headline": _rule_headline(actions, len(decisions)),
             "debug": {"decisions": decisions} if debug else None,
@@ -600,7 +606,7 @@ class FlowLedgerService:
             "edge_counts": dict(edges),
             "edge_labels": {key: WALK_EDGE_LABELS.get(key, "系统原因待补充") for key in edges},
             "stop_reason": reasons.most_common(1)[0][0] if reasons else "",
-            "stop_reason_label": REASON_LABELS.get(reasons.most_common(1)[0][0], "本轮没有继续扩展") if reasons else "本轮没有继续扩展",
+            "stop_reason_label": _reason_label(reasons.most_common(1)[0][0], "本轮没有继续扩展") if reasons else "本轮没有继续扩展",
             "headline": f"{len(actions)} 次扩展动作" if actions else "本轮没有继续扩展",
         }
 
@@ -1368,10 +1374,10 @@ def _walk_reason_label(action: dict[str, Any], reason: str) -> str:
     if gate_reason == "v4_allow_walk_denied":
         allow_reason = _text(_action_value(action, "allow_walk_reason"))
         if allow_reason and allow_reason != gate_reason:
-            return f"不适合继续向外扩展：{REASON_LABELS.get(allow_reason, '评分未达到继续扩展门槛')}"
+            return f"不适合继续向外扩展：{_reason_label(allow_reason, '评分未达到继续扩展门槛')}"
         return "不适合继续向外扩展"
     if reason:
-        return REASON_LABELS.get(reason, "这一步没有记录额外原因")
+        return _reason_label(reason, "这一步没有记录额外原因")
     return "这一步没有记录额外原因"
 
 

tests/test_v4_validator_contract.py

@@ -144,6 +144,38 @@ def test_v4_action_thresholds_reject_conflicting_action(tmp_path):
     assert "v4_action_threshold_mismatch" in _check_ids(result)
 
 
+def test_v4_validator_uses_decision_score_thresholds(tmp_path):
+    def mutate(data: dict[str, Any]) -> None:
+        decision = data["rule_decisions.jsonl"][0]
+        decision["score"] = 67.5
+        decision["scorecard"]["query_relevance_score"] = 70
+        decision["scorecard"]["platform_performance_score"] = 65
+        decision["scorecard"]["score_thresholds"] = {
+            "pool_total": 65,
+            "pool_query": 65,
+            "review_total": 55,
+            "review_query": 55,
+            "walk_query": 65,
+            "walk_platform": 65,
+            "walk_total": 65,
+        }
+        decision["decision_action"] = "ADD_TO_CONTENT_POOL"
+        decision["decision_replay_data"]["allow_walk"] = True
+        decision["decision_replay_data"]["allow_walk_reason"] = "query>=65/platform>=65/score>=65"
+        decision["decision_replay_data"]["walk_gate_snapshot"] = {
+            "query_relevance_score": 70,
+            "platform_performance_score": 65,
+            "score": 67.5,
+        }
+        _refresh_final_output_explanations(data)
+
+    runtime = _write_runtime(tmp_path, mutate)
+
+    result = validate_run(RUN_ID, runtime)
+
+    assert result["status"] == "pass"
+
+
 def test_v4_gemini_failure_requires_structured_failure_fields(tmp_path):
     def mutate(data: dict[str, Any]) -> None:
         summary = data["pattern_recall_evidence.jsonl"][0]["evidence_summary"]