Route V4 technical failures to retry state

content_agent/business_modules/learning_review.py

@@ -13,6 +13,7 @@ CURRENT_DECISION_ACTIONS = {
     "ADD_TO_CONTENT_POOL",
     "KEEP_CONTENT_FOR_REVIEW",
     "REJECT_CONTENT",
+    "TECHNICAL_RETRY_REQUIRED",
 }
 CURRENT_QUERY_EFFECT_STATUSES = {"success", "pending", "failed", "rule_blocked"}
 
@@ -167,6 +168,7 @@ def _build_metric_summary(
             "rule_blocked_query_count": status_counts["rule_blocked"],
             "pooled_decision_count": action_counts["ADD_TO_CONTENT_POOL"],
             "review_decision_count": action_counts["KEEP_CONTENT_FOR_REVIEW"],
+            "technical_retry_decision_count": action_counts["TECHNICAL_RETRY_REQUIRED"],
             "rejected_decision_count": action_counts["REJECT_CONTENT"],
         }
     )

content_agent/business_modules/result_source_lookup.py

@@ -39,6 +39,12 @@ def run(
         media_by_platform_content_id,
         paths_by_content_id,
     )
+    technical_retry_records = _build_technical_retry_records(
+        policy_run_id,
+        discovered_content_items,
+        decision_by_target_id,
+        media_by_platform_content_id,
+    )
     reject_records = _build_reject_records(
         policy_run_id,
         discovered_content_items,
@@ -82,6 +88,7 @@ def run(
         "content_assets": content_assets,
         "author_assets": author_assets,
         "review_records": review_records,
+        "technical_retry_records": technical_retry_records,
         "decision_records": [
             {
                 "decision_id": decision["decision_id"],
@@ -117,6 +124,7 @@ def run(
             "review_content_count": action_counts["KEEP_CONTENT_FOR_REVIEW"],
             "pending_content_count": 0,
             "rejected_content_count": action_counts["REJECT_CONTENT"],
+            "technical_retry_content_count": action_counts["TECHNICAL_RETRY_REQUIRED"],
             "effect_status_counts": {
                 "success": effect_status_counts["success"],
                 "pending": effect_status_counts["pending"],
@@ -260,6 +268,35 @@ def _build_reject_records(
     return reject_records
 
 
+def _build_technical_retry_records(
+    policy_run_id: str,
+    discovered_content_items: list[dict[str, Any]],
+    decision_by_target_id: dict[str, dict[str, Any]],
+    media_by_platform_content_id: dict[str, dict[str, Any]],
+) -> list[dict[str, Any]]:
+    retry_records: list[dict[str, Any]] = []
+    for item in discovered_content_items:
+        platform_content_id = item["platform_content_id"]
+        decision = decision_by_target_id[platform_content_id]
+        if decision["decision_action"] != "TECHNICAL_RETRY_REQUIRED":
+            continue
+        retry_records.append(
+            _with_v4_explanation(
+                {
+                    "decision_target_id": platform_content_id,
+                    "policy_run_id": policy_run_id,
+                    "main_decision_reason_code": decision["decision_reason_code"],
+                    "decision_id": decision["decision_id"],
+                    "technical_retry_status": "retry_required",
+                    "media_snapshot": _media_snapshot(media_by_platform_content_id.get(platform_content_id)),
+                    "source_evidence": decision["source_evidence"],
+                },
+                decision,
+            )
+        )
+    return retry_records
+
+
 def _media_snapshot(media: dict[str, Any] | None) -> dict[str, Any]:
     if not media:
         return {}
@@ -314,6 +351,7 @@ def _v4_decision_explanation(decision: dict[str, Any]) -> dict[str, Any]:
         "score_missing",
         "failure_type",
         "exception_type",
+        "error_message",
         "http_status_code",
         "retry_count",
         "final_status",

content_agent/business_modules/rule_judgment/evaluator.py

@@ -10,11 +10,13 @@ HARD_GATE_REASON_CATEGORY = "hard_gate"
 SCORE_REASON_CATEGORY = "score_or_data_failed"
 REVIEW_REASON_CATEGORY = "review_needed"
 SUCCESS_REASON_CATEGORY = "score_pass"
+TECHNICAL_REASON_CATEGORY = "technical_error"
 V4_SCORECARD_SCHEMA_VERSION = "v4_scorecard.v1"
 V4_POOL_REASON = "v4_query_and_platform_pass"
 V4_REVIEW_REASON = "v4_score_review_needed"
 V4_REJECT_REASON = "v4_query_or_score_below_threshold"
 V4_TECHNICAL_RETRY_REASON = "v4_technical_retry_needed"
+V4_TECHNICAL_RETRY_ACTION = "TECHNICAL_RETRY_REQUIRED"
 
 
 def decide(
@@ -142,7 +144,7 @@ def _decide_v4(
     scorecard = _v4_scorecard_total(bundle)
     score = scorecard.get("total_score")
     if score is None:
-        decision_action = "KEEP_CONTENT_FOR_REVIEW"
+        decision_action = V4_TECHNICAL_RETRY_ACTION
         decision_reason_code = V4_TECHNICAL_RETRY_REASON
     else:
         decision_action, decision_reason_code = _v4_decision(scorecard)
@@ -190,6 +192,7 @@ def _v4_scorecard_total(bundle: dict[str, Any]) -> dict[str, Any]:
         "missing_observable_fields": missing if isinstance(missing, list) else [],
         "total_score": total,
         "score_missing": total is None,
+        **_technical_failure_fields(bundle),
     }
 
 
@@ -212,6 +215,17 @@ def _v4_decision(scorecard: dict[str, Any]) -> tuple[str, str]:
     return "REJECT_CONTENT", V4_REJECT_REASON
 
 
+def _technical_failure_fields(bundle: dict[str, Any]) -> dict[str, Any]:
+    pattern = bundle.get("pattern_match_result")
+    if not isinstance(pattern, dict):
+        return {}
+    fields = {}
+    for field in ["failure_type", "exception_type", "http_status_code", "retry_count", "final_status"]:
+        if field in pattern:
+            fields[field] = pattern.get(field)
+    return fields
+
+
 def _v4_allow_walk(scorecard: dict[str, Any], score: Any) -> bool:
     query = scorecard.get("query_relevance_score")
     platform = scorecard.get("platform_performance_score")
@@ -514,12 +528,25 @@ def _effect_status_for_decision(
     ]
     exact = [mapping for mapping in candidates if mapping.get("reason_category") == reason_category]
     matches = exact or candidates
+    if (
+        not matches
+        and decision_action == V4_TECHNICAL_RETRY_ACTION
+        and reason_category == TECHNICAL_REASON_CATEGORY
+        and not is_hard_gate
+    ):
+        return {
+            "mapping_id": "fallback_technical_retry_failed",
+            "content_effect_status": "failed",
+            "query_effect_status": "failed",
+        }
     if not matches:
         raise ValueError(f"effect mapping not found for {decision_action}/{reason_category}")
     return sorted(matches, key=lambda mapping: (mapping.get("priority", 999), mapping.get("mapping_id", "")))[0]
 
 
 def _reason_category_for_threshold(decision_action: str, decision_reason_code: str) -> str:
+    if decision_reason_code == V4_TECHNICAL_RETRY_REASON:
+        return TECHNICAL_REASON_CATEGORY
     if decision_action == "ADD_TO_CONTENT_POOL":
         return SUCCESS_REASON_CATEGORY
     if decision_action == "KEEP_CONTENT_FOR_REVIEW":

content_agent/business_modules/run_record/recorder.py

@@ -144,6 +144,7 @@ def _build_search_clues(
         result_count = len(query_items)
         pooled_content_count = action_counts["ADD_TO_CONTENT_POOL"]
         review_content_count = action_counts["KEEP_CONTENT_FOR_REVIEW"]
+        technical_retry_content_count = action_counts["TECHNICAL_RETRY_REQUIRED"]
         pending_content_count = 0
         rejected_content_count = action_counts["REJECT_CONTENT"]
         if query_failure:
@@ -171,6 +172,7 @@ def _build_search_clues(
                     "result_count": result_count,
                     "pooled_content_count": pooled_content_count,
                     "review_content_count": review_content_count,
+                    "technical_retry_content_count": technical_retry_content_count,
                     "pending_content_count": pending_content_count,
                     "rejected_content_count": rejected_content_count,
                     "search_query_effect_status": search_query_effect_status,
@@ -338,4 +340,3 @@ def record_stage_event(
     )
     runtime.append_jsonl(run_id, "run_events.jsonl", [row])
 
-

content_agent/business_modules/run_record/validation.py

@@ -61,6 +61,7 @@ DECISION_ACTIONS = {
     "ADD_TO_CONTENT_POOL",
     "KEEP_CONTENT_FOR_REVIEW",
     "REJECT_CONTENT",
+    "TECHNICAL_RETRY_REQUIRED",
 }
 EFFECT_STATUSES = {"success", "pending", "failed", "rule_blocked"}
 WALK_STATUSES = {"success", "pending", "failed", "skipped", "rule_blocked"}
@@ -173,6 +174,12 @@ def compute_final_output_completeness(
         if record.get("decision_id") not in decision_ids:
             findings.append(f"reject_record missing decision: {record.get('decision_target_id')}")
 
+    for record in final_output.get("technical_retry_records", []):
+        if record.get("decision_id") not in decision_ids:
+            findings.append(
+                f"technical_retry_record missing decision: {record.get('decision_target_id')}"
+            )
+
     final_decision_ids = {
         record.get("decision_id") for record in final_output.get("decision_records", [])
     }
@@ -470,7 +477,7 @@ def _check_references(data: dict[str, Any], findings: list[dict[str, Any]]) -> N
                     "missing_path_ref",
                     f"review_records has unknown path_id: {path_id}",
                 )
-    for section in ["reject_records", "decision_records"]:
+    for section in ["reject_records", "technical_retry_records", "decision_records"]:
         for row in final_output.get(section, []):
             if row.get("decision_id") not in decision_ids:
                 _fail(
@@ -560,6 +567,13 @@ def _check_source_evidence(data: dict[str, Any], findings: list[dict[str, Any]])
             evidence_pack,
             f"review {record.get('platform_content_id')}",
         )
+    for record in final_output.get("technical_retry_records", []):
+        _check_one_source_evidence(
+            findings,
+            record.get("source_evidence") or {},
+            evidence_pack,
+            f"technical_retry {record.get('decision_target_id')}",
+        )
     for record in final_output.get("decision_records", []):
         _check_one_source_evidence(
             findings,
@@ -788,6 +802,11 @@ def _check_summary(data: dict[str, Any], findings: list[dict[str, Any]]) -> None
         "pending_content_count": 0,
         "rejected_content_count": action_counts["REJECT_CONTENT"],
     }
+    if (
+        action_counts["TECHNICAL_RETRY_REQUIRED"]
+        or "technical_retry_content_count" in summary
+    ):
+        expected["technical_retry_content_count"] = action_counts["TECHNICAL_RETRY_REQUIRED"]
     for field, value in expected.items():
         if summary.get(field) != value:
             _fail(
@@ -801,7 +820,10 @@ def _check_summary(data: dict[str, Any], findings: list[dict[str, Any]]) -> None
         clue_counts["ADD_TO_CONTENT_POOL"] += clue.get("pooled_content_count", 0)
         clue_counts["KEEP_CONTENT_FOR_REVIEW"] += clue.get("review_content_count", 0)
         clue_counts["REJECT_CONTENT"] += clue.get("rejected_content_count", 0)
+        clue_counts["TECHNICAL_RETRY_REQUIRED"] += clue.get("technical_retry_content_count", 0)
     for action, count in action_counts.items():
+        if action == "TECHNICAL_RETRY_REQUIRED" and not count:
+            continue
         if clue_counts[action] < count:
             _fail(
                 findings,

content_agent/business_modules/walk_strategy.py

@@ -30,6 +30,14 @@ def _action_for_decision(decision_action: str) -> dict[str, str]:
             "walk_status": "success",
             "budget_tier": "low_budget",
         }
+    if decision_action == "TECHNICAL_RETRY_REQUIRED":
+        return {
+            "edge_id": "path_stop",
+            "edge_type": "terminal",
+            "walk_action": "stop_path",
+            "walk_status": "failed",
+            "budget_tier": "stop",
+        }
     return {
         "edge_id": "path_stop",
         "edge_type": "terminal",

content_agent/integrations/walk_graph_json.py

@@ -18,7 +18,12 @@ WALK_POLICY_PATH = Path("tech_documents/数据接口与来源/walk_policy.json")
 PROFILE_DIR = Path("tech_documents/数据接口与来源/platform_profiles")
 EDGE_CATALOG_PATH = Path("product_documents/抖音游走策略/douyin_walk_strategy.v1.json")
 
-PERMISSION_ACTIONS = ["ADD_TO_CONTENT_POOL", "KEEP_CONTENT_FOR_REVIEW", "REJECT_CONTENT"]
+PERMISSION_ACTIONS = [
+    "ADD_TO_CONTENT_POOL",
+    "KEEP_CONTENT_FOR_REVIEW",
+    "TECHNICAL_RETRY_REQUIRED",
+    "REJECT_CONTENT",
+]
 _PERMISSION_META_KEYS = {"_meaning", "_provenance", "note", "tbd"}
 
 

content_agent/models.py

@@ -8,6 +8,7 @@ DecisionAction = Literal[
     "ADD_TO_CONTENT_POOL",
     "KEEP_CONTENT_FOR_REVIEW",
     "REJECT_CONTENT",
+    "TECHNICAL_RETRY_REQUIRED",
 ]
 SearchQueryEffectStatus = Literal["success", "pending", "failed", "rule_blocked"]
 

product_documents/规则包/douyin_rule_packs.v1.json

@@ -2,7 +2,7 @@
   "schema_version": "rule_pack.v1",
   "package_id": "douyin_rule_packs_v1",
   "package_name": "抖音规则包 V4",
-  "description": "核心对象说明：EvidenceBundle 是判断证据包，RuleDecision 是规则判断结果，platform_content_id 是平台内容 ID，platform_author_id 是抖音作者 ID，run_id 是本次运行 ID，policy_run_id 是本次策略执行 ID，record_schema_version 是运行记录结构版本，source_evidence 是来源证据，search_query_effect_status 是搜索词效果状态。Content V4 runtime 只使用 success / pending / failed / rule_blocked 四类效果状态。ADD_TO_CONTENT_POOL 是入池，KEEP_CONTENT_FOR_REVIEW 是待复看并映射 pending，REJECT_CONTENT 是淘汰。",
+  "description": "核心对象说明：EvidenceBundle 是判断证据包，RuleDecision 是规则判断结果，platform_content_id 是平台内容 ID，platform_author_id 是抖音作者 ID，run_id 是本次运行 ID，policy_run_id 是本次策略执行 ID，record_schema_version 是运行记录结构版本，source_evidence 是来源证据，search_query_effect_status 是搜索词效果状态。Content V4 runtime 只使用 success / pending / failed / rule_blocked 四类效果状态。ADD_TO_CONTENT_POOL 是入池，KEEP_CONTENT_FOR_REVIEW 是待复看并映射 pending，TECHNICAL_RETRY_REQUIRED 是技术链路失败需重试并映射 failed，REJECT_CONTENT 是淘汰。",
   "updated_at": "2026-06-15",
   "strategy_binding": {
     "strategy_id": "douyin_content_find_v4",
@@ -41,11 +41,13 @@
       "allowed_actions": [
         "ADD_TO_CONTENT_POOL",
         "KEEP_CONTENT_FOR_REVIEW",
+        "TECHNICAL_RETRY_REQUIRED",
         "REJECT_CONTENT"
       ],
       "action_notes": {
         "ADD_TO_CONTENT_POOL": "入池成功，映射 content_effect_status=success。",
         "KEEP_CONTENT_FOR_REVIEW": "待人工复看，映射 content_effect_status=pending，不是弱成功。",
+        "TECHNICAL_RETRY_REQUIRED": "下载、压缩、Gemini/OpenRouter 等技术链路失败，映射 content_effect_status=failed，不代表内容待人工复看。",
         "REJECT_CONTENT": "淘汰；按原因映射 failed 或 rule_blocked。"
       }
     },
@@ -188,11 +190,11 @@
         ],
         "missing_policy": "fail_hard_gate",
         "score_missing_policy": {
-          "decision_action": "KEEP_CONTENT_FOR_REVIEW",
+          "decision_action": "TECHNICAL_RETRY_REQUIRED",
           "decision_reason_code": "v4_technical_retry_needed",
-          "effect_status": "pending",
+          "effect_status": "failed",
           "priority": 5,
-          "notes": "技术失败或 V4 分数缺失进入待复看/待补跑,不按内容质量淘汰。"
+          "notes": "技术失败或 V4 分数缺失进入技术重试,不按内容质量淘汰,也不计入人工待复看。"
         }
       },
       "hard_gates": [
@@ -280,9 +282,9 @@
             "op": "eq",
             "value": "failed"
           },
-          "decision_action": "KEEP_CONTENT_FOR_REVIEW",
+          "decision_action": "TECHNICAL_RETRY_REQUIRED",
           "decision_reason_code": "v4_technical_retry_needed",
-          "severity": "review",
+          "severity": "technical",
           "stop_scoring": true,
           "priority": 15
         }
@@ -291,9 +293,9 @@
         "schema_version": "v4_scorecard.v1",
         "total_score": 100,
         "score_missing_policy": {
-          "decision_action": "KEEP_CONTENT_FOR_REVIEW",
+          "decision_action": "TECHNICAL_RETRY_REQUIRED",
           "decision_reason_code": "v4_technical_retry_needed",
-          "effect_status": "pending",
+          "effect_status": "failed",
           "priority": 5
         },
         "dimensions": [
@@ -506,6 +508,30 @@
       "enabled": true,
       "notes": "画像缺失等 hard gate 直接给待复看时，内容与 query 状态都记 pending。"
     },
+    {
+      "mapping_id": "map_technical_retry_failed",
+      "target_level": "content",
+      "decision_action": "TECHNICAL_RETRY_REQUIRED",
+      "reason_category": "technical_error",
+      "is_hard_gate": false,
+      "content_effect_status": "failed",
+      "query_effect_status": "failed",
+      "priority": 28,
+      "enabled": true,
+      "notes": "下载、压缩、Gemini/OpenRouter 等技术链路失败需要重试,不计入待复看。"
+    },
+    {
+      "mapping_id": "map_technical_retry_failed_hard_gate",
+      "target_level": "content",
+      "decision_action": "TECHNICAL_RETRY_REQUIRED",
+      "reason_category": "hard_gate",
+      "is_hard_gate": true,
+      "content_effect_status": "failed",
+      "query_effect_status": "failed",
+      "priority": 29,
+      "enabled": true,
+      "notes": "judge_failed hard gate 触发的技术失败同样计为 failed,避免污染待复看。"
+    },
     {
       "mapping_id": "map_reject_rule_blocked",
       "target_level": "content",

tech_documents/数据接口与来源/walk_policy.json

@@ -25,6 +25,7 @@
     "_provenance": "现状行为来自 walk_engine v1 + E2E 实测(KEEP 内容 author_to_works success(low_budget));M4 拍板 2026-06-11",
     "ADD_TO_CONTENT_POOL":     { "author_to_works": "allow",            "video_to_hashtag": "allow", "decision_to_asset": "allow" },
     "KEEP_CONTENT_FOR_REVIEW": { "author_to_works": "allow_low_budget", "video_to_hashtag": { "value": "deny", "tbd": false, "note": "拍板 2026-06-11:从严 deny(v1 等价,防漂移)" }, "decision_to_asset": "deny" },
+    "TECHNICAL_RETRY_REQUIRED": { "author_to_works": "deny", "video_to_hashtag": "deny", "decision_to_asset": "deny", "note": "技术链路失败不触发业务游走或资产沉淀" },
     "REJECT_CONTENT":          { "author_to_works": "deny", "video_to_hashtag": "deny", "decision_to_asset": "deny", "note": "即 path_stop:rule_blocked/failed 不开任何出边" }
   },
   "v4_walk_gate": {

tests/test_replay_gemini_seam.py

@@ -43,18 +43,19 @@ def test_replay_review_stub_scores_by_relevance(tmp_path):
     ]
 
 
-def test_replay_fail_stub_routes_to_judge_failed_review(tmp_path):
+def test_replay_fail_stub_routes_to_technical_retry(tmp_path):
     artifacts = replay_case(
         "real_id45",
         runtime_root=tmp_path / "rt",
         gemini_video_client=FakeGeminiVideoClient(default_result=fake_gemini_fail()),
     )
     assert artifacts.state["status"] == "success"
-    assert artifacts.summary["review_content_count"] == 4
+    assert artifacts.summary["review_content_count"] == 0
+    assert artifacts.summary["technical_retry_content_count"] == 4
     assert artifacts.summary["pooled_content_count"] == 0
     assert artifacts.summary["rejected_content_count"] == 0
     assert all(
-        d["decision_action"] == "KEEP_CONTENT_FOR_REVIEW"
+        d["decision_action"] == "TECHNICAL_RETRY_REQUIRED"
         and d["decision_reason_code"] == "v4_technical_retry_needed"
         and d["decision_replay_data"]["allow_walk"] is False
         for d in artifacts.decisions

tests/test_rule_decision_effect_status.py

@@ -25,9 +25,10 @@ def test_v4_effect_status_mapping_for_pool_review_reject_technical_and_blocked()
     assert failed["decision_replay_data"]["effect_mapping_id"] == "map_reject_failed"
 
     technical = decide("run_1", "policy_1", 4, _bundle(query=None, platform=70), policy)
+    assert technical["decision_action"] == "TECHNICAL_RETRY_REQUIRED"
     assert technical["decision_reason_code"] == "v4_technical_retry_needed"
-    assert technical["search_query_effect_status"] == "pending"
-    assert technical["decision_replay_data"]["effect_mapping_id"] == "map_keep_for_review_pending"
+    assert technical["search_query_effect_status"] == "failed"
+    assert technical["decision_replay_data"]["effect_mapping_id"] == "map_technical_retry_failed"
 
     blocked_policy = deepcopy(policy)
     blocked_policy["rule_pack"]["hard_gates"] = [

tests/test_rule_judgment_hard_gates.py

@@ -59,14 +59,14 @@ def test_legacy_senior_fit_field_no_longer_triggers_hard_gate(tmp_path):
 
 
 def test_hard_gate_action_is_taken_from_rule_config(tmp_path):
-    # The judge_failed gate defaults to KEEP_CONTENT_FOR_REVIEW; flipping its config to
+    # The judge_failed gate defaults to TECHNICAL_RETRY_REQUIRED; flipping its config to
     # REJECT_CONTENT must change the decision purely via config, with no code special-case.
     state = _state(tmp_path)
     bundle = deepcopy(state["evidence_bundles"][0])
     bundle["pattern_match_result"]["judge_status"] = "failed"
 
     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_action"] == "TECHNICAL_RETRY_REQUIRED"
 
     flipped = deepcopy(state["policy_bundle"])
     for gate in flipped["rule_pack"]["hard_gates"]:
@@ -78,21 +78,20 @@ def test_hard_gate_action_is_taken_from_rule_config(tmp_path):
     assert decision["search_query_effect_status"] == "rule_blocked"
 
 
-def test_judge_failed_review_is_config_driven_not_code_special_case(tmp_path):
-    # M3: the review/pending hard gate is now judge_failed (Gemini technical failure parks
-    # content for human review). This is config-driven, not a code special-case.
+def test_judge_failed_technical_retry_is_config_driven_not_code_special_case(tmp_path):
+    # Gemini technical failure is config-driven, not a code special-case.
     state = _state(tmp_path)
     bundle = deepcopy(state["evidence_bundles"][0])
     bundle["pattern_match_result"]["judge_status"] = "failed"
 
     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_action"] == "TECHNICAL_RETRY_REQUIRED"
     assert decision["decision_reason_code"] == "v4_technical_retry_needed"
-    assert decision["search_query_effect_status"] == "pending"
+    assert decision["search_query_effect_status"] == "failed"
     assert decision["triggered_blocking_rules"] == ["judge_failed"]
     assert decision["score"] is None
-    assert decision["decision_replay_data"]["effect_mapping_id"] == "map_keep_for_review_pending_hard_gate"
+    assert decision["decision_replay_data"]["effect_mapping_id"] == "map_technical_retry_failed_hard_gate"
 
 
 def test_legacy_low_confidence_field_no_longer_triggers_hard_gate(tmp_path):

tests/test_rule_judgment_scorecard.py

@@ -44,14 +44,14 @@ def test_v4_allow_walk_requires_platform_65():
     assert decision["decision_replay_data"]["allow_walk"] is False
 
 
-def test_v4_missing_score_routes_to_technical_retry_review():
+def test_v4_missing_score_routes_to_technical_retry():
     decision = decide("run_1", "policy_1", 1, _bundle(query=None, platform=70), _policy())
 
     assert decision["score"] is None
     assert decision["scorecard"]["score_missing"] is True
-    assert decision["decision_action"] == "KEEP_CONTENT_FOR_REVIEW"
+    assert decision["decision_action"] == "TECHNICAL_RETRY_REQUIRED"
     assert decision["decision_reason_code"] == "v4_technical_retry_needed"
-    assert decision["search_query_effect_status"] == "pending"
+    assert decision["search_query_effect_status"] == "failed"
     assert decision["decision_replay_data"]["allow_walk"] is False
 
 
@@ -148,6 +148,16 @@ def _policy():
                     "priority": 40,
                     "enabled": True,
                 },
+                {
+                    "mapping_id": "map_technical_retry_failed",
+                    "target_level": "content",
+                    "decision_action": "TECHNICAL_RETRY_REQUIRED",
+                    "reason_category": "technical_error",
+                    "is_hard_gate": False,
+                    "content_effect_status": "failed",
+                    "priority": 30,
+                    "enabled": True,
+                },
             ],
         }
     )

tests/test_rule_pack_reading.py

@@ -67,9 +67,9 @@ def test_missing_score_uses_rule_pack_missing_policy(tmp_path):
     bundle["content_engagement_metrics"]["platform_performance"]["platform_performance_score"] = None
 
     decision = decide(run_id, state["policy_run_id"], 1, bundle, state["policy_bundle"])
-    assert decision["decision_action"] == "KEEP_CONTENT_FOR_REVIEW"
+    assert decision["decision_action"] == "TECHNICAL_RETRY_REQUIRED"
     assert decision["decision_reason_code"] == "v4_technical_retry_needed"
-    assert decision["search_query_effect_status"] == "pending"
+    assert decision["search_query_effect_status"] == "failed"
     assert decision["scorecard"]["score_missing"] is True
 
 

tests/test_v4_m3_scoring_replay.py

@@ -52,7 +52,7 @@ def test_v4_m3_scoring_replay_produces_v4_runtime_contract(tmp_path):
         assert "platform_heat" not in scorecard
 
 
-def test_v4_m3_scoring_replay_keeps_technical_failure_in_review(tmp_path):
+def test_v4_m3_scoring_replay_routes_technical_failure_to_retry(tmp_path):
     artifacts = replay_case(
         "real_id45",
         runtime_root=tmp_path / "runtime",
@@ -61,8 +61,10 @@ def test_v4_m3_scoring_replay_keeps_technical_failure_in_review(tmp_path):
 
     assert artifacts.summary["pooled_content_count"] == 0
     assert artifacts.summary["rejected_content_count"] == 0
-    assert artifacts.summary["review_content_count"] == len(artifacts.decisions)
+    assert artifacts.summary["review_content_count"] == 0
+    assert artifacts.summary["technical_retry_content_count"] == len(artifacts.decisions)
     assert {d["decision_reason_code"] for d in artifacts.decisions} == {"v4_technical_retry_needed"}
+    assert {d["decision_action"] for d in artifacts.decisions} == {"TECHNICAL_RETRY_REQUIRED"}
     assert all(d["decision_replay_data"]["allow_walk"] is False for d in artifacts.decisions)
     assert all(d["scorecard"]["schema_version"] == "v4_scorecard.v1" for d in artifacts.decisions)