feat: add V4 M5 explanations and M6 real acceptance

content_agent/api.py

@@ -168,7 +168,8 @@ _PLATFORM_PROFILE_DIR = "tech_documents/数据接口与来源/platform_profiles"
 
 @app.get("/config/platforms", response_model=PlatformCatalogResponse)
 def get_config_platforms() -> PlatformCatalogResponse:
-    # 平台展示目录:从 platform_profiles/*.json 派生每平台的 label + 真实可得互动指标(heat 字段)。
+    # 平台展示目录:从 platform_profiles/*.json 派生平台 label 与可观测字段。
+    # heat_fields 是 V3 历史展示字段;observable_fields 才是 V4 平台表现解释字段。
     # 前端按此渲染平台名与互动数据——加平台只改 profile JSON,前端零改动。
     from pathlib import Path
 
@@ -187,11 +188,17 @@ def get_config_platforms() -> PlatformCatalogResponse:
             for sig in ((data.get("heat") or {}).get("signals") or [])
             if isinstance(sig, dict) and sig.get("field")
         ]
+        observable_fields = [
+            str(item["field"])
+            for item in (data.get("observable_fields") or [])
+            if isinstance(item, dict) and item.get("field")
+        ]
         catalog[platform] = PlatformDescriptor(
             platform=platform,
             label=str(data.get("platform_label") or platform),
             status=data.get("status"),
             heat_fields=heat_fields,
+            observable_fields=observable_fields,
         )
     return PlatformCatalogResponse(platforms=catalog)
 

content_agent/business_modules/learning_review.py

@@ -35,6 +35,13 @@ def run(
     rule_review = _build_rule_review(decisions)
     performance_feedback = _build_performance_feedback_summary(performance_rows)
     productive_paths = _build_productive_paths(source_path_records)
+    walk_review = _build_walk_review(walk_actions)
+    v4_summary = _build_v4_summary(
+        decisions,
+        walk_actions,
+        final_output,
+        source_path_records,
+    )
     search_clue_assets, search_clue_evidence = _build_search_clue_asset_promotions(
         run_id,
         policy_run_id,
@@ -70,9 +77,10 @@ def run(
         "metric_summary": _build_metric_summary(final_output, search_clues, decisions),
         "query_review": query_review,
         "rule_review": rule_review,
-        "walk_review": _build_walk_review(walk_actions),
+        "walk_review": walk_review,
         "asset_review": _build_asset_review(final_output, search_clue_assets),
         "performance_feedback": performance_feedback,
+        "v4_summary": v4_summary,
         "recommendations": recommendations,
         "decision_distribution": rule_review["decision_distribution"],
         "effective_search_queries": [
@@ -219,10 +227,187 @@ def _build_rule_review(decisions: list[dict[str, Any]]) -> dict[str, Any]:
 def _build_walk_review(walk_actions: list[dict[str, Any]]) -> dict[str, Any]:
     status_counts = Counter(action.get("walk_status") for action in walk_actions)
     edge_counts = Counter(action.get("edge_id") for action in walk_actions)
+    gate_review = _walk_gate_review(walk_actions)
     return {
         "walk_action_count": len(walk_actions),
         "walk_status_distribution": dict(status_counts),
         "edge_distribution": dict(edge_counts),
+        "v4_gate_distribution": gate_review["v4_gate_distribution"],
+        "deny_reasons": gate_review["deny_reasons"],
+    }
+
+
+def _build_v4_summary(
+    decisions: list[dict[str, Any]],
+    walk_actions: list[dict[str, Any]],
+    final_output: dict[str, Any],
+    source_path_records: list[dict[str, Any]],
+) -> dict[str, Any]:
+    v4_decisions = [decision for decision in decisions if _is_v4_decision(decision)]
+    walk_gate_review = _walk_gate_review(walk_actions)
+    return {
+        "schema_version": "v4_strategy_review_summary.v1",
+        "summary_status": "available" if v4_decisions else "missing",
+        "v4_decision_count": len(v4_decisions),
+        "content_asset_count": len(final_output.get("content_assets") or []),
+        "source_path_record_count": len(source_path_records),
+        "decision_distribution": dict(
+            Counter(decision.get("decision_action") for decision in v4_decisions)
+        ),
+        "score_buckets": _score_buckets(v4_decisions),
+        "score_ranges": _score_ranges(v4_decisions),
+        "missing_observable_fields": _missing_observable_field_counts(v4_decisions),
+        "technical_retry": _technical_retry_summary(v4_decisions),
+        "allow_walk_distribution": _allow_walk_distribution(v4_decisions),
+        "walk_gate_review": walk_gate_review,
+    }
+
+
+def _is_v4_decision(decision: dict[str, Any]) -> bool:
+    scorecard = decision.get("scorecard") or {}
+    return isinstance(scorecard, dict) and scorecard.get("schema_version") == "v4_scorecard.v1"
+
+
+def _score_bucket(decision: dict[str, Any]) -> str:
+    if decision.get("decision_reason_code") == "v4_technical_retry_needed":
+        return "technical_retry"
+    action = decision.get("decision_action")
+    if action == "ADD_TO_CONTENT_POOL":
+        return "pool"
+    if action == "KEEP_CONTENT_FOR_REVIEW":
+        return "review"
+    if action == "REJECT_CONTENT":
+        return "reject"
+    return "unknown"
+
+
+def _score_buckets(decisions: list[dict[str, Any]]) -> dict[str, int]:
+    counts = Counter(_score_bucket(decision) for decision in decisions)
+    return {
+        "pool": counts["pool"],
+        "review": counts["review"],
+        "reject": counts["reject"],
+        "technical_retry": counts["technical_retry"],
+        "unknown": counts["unknown"],
+    }
+
+
+def _score_ranges(decisions: list[dict[str, Any]]) -> dict[str, dict[str, float | int | None]]:
+    return {
+        "query_relevance_score": _range_summary(
+            (decision.get("scorecard") or {}).get("query_relevance_score")
+            for decision in decisions
+        ),
+        "platform_performance_score": _range_summary(
+            (decision.get("scorecard") or {}).get("platform_performance_score")
+            for decision in decisions
+        ),
+        "score": _range_summary(decision.get("score") for decision in decisions),
+    }
+
+
+def _range_summary(values: Any) -> dict[str, float | int | None]:
+    numeric_values = [
+        float(value)
+        for value in values
+        if isinstance(value, (int, float)) and not isinstance(value, bool)
+    ]
+    if not numeric_values:
+        return {"min": None, "max": None, "avg": None}
+    return {
+        "min": min(numeric_values),
+        "max": max(numeric_values),
+        "avg": round(sum(numeric_values) / len(numeric_values), 4),
+    }
+
+
+def _missing_observable_field_counts(decisions: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    counts: Counter[tuple[str, str]] = Counter()
+    for decision in decisions:
+        scorecard = decision.get("scorecard") or {}
+        for item in scorecard.get("missing_observable_fields") or []:
+            if isinstance(item, dict):
+                field = str(item.get("field") or "unknown")
+                reason = str(item.get("reason") or "unknown")
+            else:
+                field = str(item)
+                reason = "unknown"
+            counts[(field, reason)] += 1
+    return [
+        {"field": field, "reason": reason, "count": count}
+        for (field, reason), count in sorted(counts.items())
+    ]
+
+
+def _technical_retry_summary(decisions: list[dict[str, Any]]) -> dict[str, Any]:
+    retry_decisions = [
+        decision
+        for decision in decisions
+        if decision.get("decision_reason_code") == "v4_technical_retry_needed"
+    ]
+    failure_types: Counter[str] = Counter()
+    for decision in retry_decisions:
+        scorecard = decision.get("scorecard") or {}
+        replay_data = decision.get("decision_replay_data") or {}
+        failure_type = (
+            scorecard.get("failure_type")
+            or replay_data.get("failure_type")
+            or scorecard.get("final_status")
+            or "unknown"
+        )
+        failure_types[str(failure_type)] += 1
+    return {
+        "count": len(retry_decisions),
+        "failure_types": dict(failure_types),
+    }
+
+
+def _allow_walk_distribution(decisions: list[dict[str, Any]]) -> dict[str, int]:
+    counts = {"allowed": 0, "denied": 0, "missing": 0}
+    for decision in decisions:
+        replay_data = decision.get("decision_replay_data") or {}
+        if replay_data.get("allow_walk") is True:
+            counts["allowed"] += 1
+        elif replay_data.get("allow_walk") is False:
+            counts["denied"] += 1
+        else:
+            counts["missing"] += 1
+    return counts
+
+
+def _walk_gate_review(walk_actions: list[dict[str, Any]]) -> dict[str, Any]:
+    gate_counts = {"allowed": 0, "denied": 0, "missing": 0}
+    deny_reasons: Counter[str] = Counter()
+    allowed_success_count = 0
+    denied_skip_count = 0
+    for action in walk_actions:
+        raw_payload = action.get("raw_payload") or {}
+        status = raw_payload.get("walk_gate_status")
+        if status == "allowed" or raw_payload.get("allow_walk") is True:
+            gate_counts["allowed"] += 1
+            if action.get("walk_status") == "success":
+                allowed_success_count += 1
+        elif status == "denied" or raw_payload.get("allow_walk") is False:
+            gate_counts["denied"] += 1
+            reason = (
+                raw_payload.get("walk_gate_reason_code")
+                or action.get("reason_code")
+                or "unknown"
+            )
+            deny_reasons[str(reason)] += 1
+            if action.get("walk_status") in {"skipped", "failed", "rule_blocked"}:
+                denied_skip_count += 1
+        else:
+            gate_counts["missing"] += 1
+    return {
+        "walk_action_count": len(walk_actions),
+        "v4_gate_distribution": gate_counts,
+        "allowed_success_count": allowed_success_count,
+        "denied_skip_count": denied_skip_count,
+        "deny_reasons": [
+            {"reason_code": reason, "count": count}
+            for reason, count in deny_reasons.most_common()
+        ],
     }
 
 

content_agent/business_modules/m6_acceptance_report.py

@@ -0,0 +1,326 @@
+from __future__ import annotations
+
+from collections import Counter
+from typing import Any, Iterable
+
+from content_agent.dashboard_service import _timeline_summary
+from content_agent.interfaces import RuntimeStore
+
+
+SCHEMA_VERSION = "v4_m6_acceptance_report.v1"
+
+REPORT_INPUT_FILES = [
+    "run_events.jsonl",
+    "pattern_recall_evidence.jsonl",
+    "rule_decisions.jsonl",
+    "walk_actions.jsonl",
+    "final_output.json",
+    "strategy_review.json",
+]
+
+
+def build_report(
+    run_id: str,
+    runtime: RuntimeStore,
+    *,
+    platform: str | None = None,
+    platform_mode: str | None = None,
+) -> dict[str, Any]:
+    run_events = _read_jsonl(runtime, run_id, "run_events.jsonl")
+    evidence_rows = _read_jsonl(runtime, run_id, "pattern_recall_evidence.jsonl")
+    decisions = _read_jsonl(runtime, run_id, "rule_decisions.jsonl")
+    walk_actions = _read_jsonl(runtime, run_id, "walk_actions.jsonl")
+    source_paths = _read_jsonl(runtime, run_id, "source_path_records.jsonl")
+    final_output = _read_json(runtime, run_id, "final_output.json")
+    strategy_review = _read_json(runtime, run_id, "strategy_review.json")
+
+    return {
+        "schema_version": SCHEMA_VERSION,
+        "run_id": run_id,
+        "policy_run_id": _policy_run_id(final_output, decisions, run_events),
+        "platform": platform or _first_value(decisions, "platform"),
+        "platform_mode": platform_mode,
+        "generated_from": {
+            "runtime_files": {
+                filename: bool(_file_exists(runtime, run_id, filename))
+                for filename in REPORT_INPUT_FILES
+            }
+        },
+        "gemini_summary": _gemini_summary(run_events, evidence_rows),
+        "performance": _performance_summary(run_events, walk_actions, source_paths),
+        "score_calibration": _score_calibration(decisions),
+        "observability": _observability_summary(decisions, walk_actions, final_output),
+        "strategy_review_summary": _strategy_review_summary(strategy_review),
+        "known_gaps": [
+            {
+                "gap_id": "tag_relevance_not_production",
+                "description": (
+                    "V4 plan mentions tag Gemini relevance, but current M4/M6 scope does "
+                    "not implement it as production semantics."
+                ),
+                "severity": "planning_gap",
+                "blocking_m6_acceptance": False,
+            }
+        ],
+    }
+
+
+def _read_json(runtime: RuntimeStore, run_id: str, filename: str) -> dict[str, Any]:
+    try:
+        return runtime.read_json(run_id, filename)
+    except FileNotFoundError:
+        return {}
+
+
+def _read_jsonl(runtime: RuntimeStore, run_id: str, filename: str) -> list[dict[str, Any]]:
+    try:
+        return runtime.read_jsonl(run_id, filename)
+    except FileNotFoundError:
+        return []
+
+
+def _file_exists(runtime: RuntimeStore, run_id: str, filename: str) -> bool:
+    try:
+        return bool(runtime.file_status(run_id).get(filename))
+    except Exception:
+        try:
+            if filename.endswith(".jsonl"):
+                return bool(runtime.read_jsonl(run_id, filename))
+            return bool(runtime.read_json(run_id, filename))
+        except Exception:
+            return False
+
+
+def _policy_run_id(
+    final_output: dict[str, Any],
+    decisions: list[dict[str, Any]],
+    run_events: list[dict[str, Any]],
+) -> str | None:
+    return (
+        final_output.get("policy_run_id")
+        or _first_value(decisions, "policy_run_id")
+        or _first_value(run_events, "policy_run_id")
+    )
+
+
+def _first_value(rows: list[dict[str, Any]], key: str) -> Any:
+    for row in rows:
+        value = row.get(key)
+        if value not in (None, ""):
+            return value
+    return None
+
+
+def _gemini_summary(
+    run_events: list[dict[str, Any]],
+    evidence_rows: list[dict[str, Any]],
+) -> dict[str, Any]:
+    quota_events = [
+        row
+        for row in run_events
+        if row.get("event_type") == "gemini_quota_exhausted"
+    ]
+    latest_quota = quota_events[-1] if quota_events else {}
+    quota_payload = latest_quota.get("raw_payload") or {}
+    summaries = [
+        row.get("evidence_summary") or row.get("raw_payload") or {}
+        for row in evidence_rows
+    ]
+    failure_type_counts: Counter[str] = Counter()
+    http_status_counts: Counter[str] = Counter()
+    retry_rescued_count = 0
+    final_failed_count = 0
+    successful_count = 0
+
+    for summary in summaries:
+        final_status = str(summary.get("final_status") or "")
+        failure_type = summary.get("failure_type")
+        if failure_type:
+            failure_type_counts[str(failure_type)] += 1
+        http_status = summary.get("http_status_code")
+        if http_status not in (None, ""):
+            http_status_counts[str(http_status)] += 1
+        retry_count = _int_value(summary.get("retry_count"))
+        if final_status in {"ok", "success"}:
+            successful_count += 1
+            if retry_count > 0:
+                retry_rescued_count += 1
+        elif final_status:
+            final_failed_count += 1
+
+    used = quota_payload.get("used")
+    if used is None:
+        used = len(summaries)
+    return {
+        "used": used,
+        "cap": quota_payload.get("cap"),
+        "quota_exhausted": bool(quota_events),
+        "successful_count": successful_count,
+        "failure_type_counts": dict(failure_type_counts),
+        "http_status_counts": dict(http_status_counts),
+        "retry_rescued_count": retry_rescued_count,
+        "final_failed_count": final_failed_count,
+    }
+
+
+def _performance_summary(
+    run_events: list[dict[str, Any]],
+    walk_actions: list[dict[str, Any]],
+    source_paths: list[dict[str, Any]],
+) -> dict[str, Any]:
+    timeline = _timeline_summary(run_events, walk_actions, source_paths)
+    return {
+        "total_duration_ms": timeline.get("total_duration_ms"),
+        "stage_duration_ms": timeline.get("stage_duration_ms", {}),
+        "platform_query_failure_count": _platform_query_failure_count(run_events),
+        "platform_rate_limited_count": timeline.get("platform_rate_limited_count", 0),
+        "error_counts": timeline.get("error_counts", {}),
+        "walk_status_counts": timeline.get("walk_status_counts", {}),
+    }
+
+
+def _platform_query_failure_count(run_events: list[dict[str, Any]]) -> int:
+    return sum(1 for row in run_events if row.get("event_type") == "platform_query_failed")
+
+
+def _score_calibration(decisions: list[dict[str, Any]]) -> dict[str, Any]:
+    return {
+        "decision_count": len(decisions),
+        "score_ranges": {
+            "query_relevance_score": _range_summary(
+                (row.get("scorecard") or {}).get("query_relevance_score")
+                for row in decisions
+            ),
+            "platform_performance_score": _range_summary(
+                (row.get("scorecard") or {}).get("platform_performance_score")
+                for row in decisions
+            ),
+            "score": _range_summary(row.get("score") for row in decisions),
+        },
+        "threshold_edge_samples": _threshold_edge_samples(decisions),
+        "decision_action_counts": dict(Counter(row.get("decision_action") for row in decisions)),
+    }
+
+
+def _range_summary(values: Iterable[Any]) -> dict[str, float | None]:
+    numeric = [
+        float(value)
+        for value in values
+        if isinstance(value, (int, float)) and not isinstance(value, bool)
+    ]
+    if not numeric:
+        return {"min": None, "max": None, "avg": None}
+    return {
+        "min": min(numeric),
+        "max": max(numeric),
+        "avg": round(sum(numeric) / len(numeric), 4),
+    }
+
+
+def _threshold_edge_samples(decisions: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    samples = []
+    for row in decisions:
+        score = row.get("score")
+        if not isinstance(score, (int, float)) or isinstance(score, bool):
+            continue
+        if 53 <= score <= 57 or 68 <= score <= 72:
+            samples.append(
+                {
+                    "decision_id": row.get("decision_id"),
+                    "decision_target_id": row.get("decision_target_id"),
+                    "score": score,
+                    "decision_action": row.get("decision_action"),
+                    "decision_reason_code": row.get("decision_reason_code"),
+                }
+            )
+    return samples[:20]
+
+
+def _observability_summary(
+    decisions: list[dict[str, Any]],
+    walk_actions: list[dict[str, Any]],
+    final_output: dict[str, Any],
+) -> dict[str, Any]:
+    return {
+        "missing_observable_fields": _missing_observable_fields(decisions),
+        "allow_walk_distribution": _allow_walk_distribution(decisions),
+        "walk_deny_reasons": _walk_deny_reasons(walk_actions),
+        "author_asset_count": len(final_output.get("author_assets") or []),
+    }
+
+
+def _missing_observable_fields(decisions: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    counts: Counter[tuple[str, str]] = Counter()
+    for row in decisions:
+        for item in (row.get("scorecard") or {}).get("missing_observable_fields") or []:
+            if isinstance(item, dict):
+                field = str(item.get("field") or "unknown")
+                reason = str(item.get("reason") or item.get("missing_type") or "unknown")
+            else:
+                field = str(item)
+                reason = "unknown"
+            counts[(field, reason)] += 1
+    return [
+        {"field": field, "reason": reason, "count": count}
+        for (field, reason), count in sorted(counts.items())
+    ]
+
+
+def _allow_walk_distribution(decisions: list[dict[str, Any]]) -> dict[str, int]:
+    counts = Counter()
+    for row in decisions:
+        replay_data = row.get("decision_replay_data") or {}
+        if replay_data.get("allow_walk") is True:
+            counts["allowed"] += 1
+        elif replay_data.get("allow_walk") is False:
+            counts["denied"] += 1
+        else:
+            counts["missing"] += 1
+    return {
+        "allowed": counts["allowed"],
+        "denied": counts["denied"],
+        "missing": counts["missing"],
+    }
+
+
+def _walk_deny_reasons(walk_actions: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    counts: Counter[str] = Counter()
+    for row in walk_actions:
+        if row.get("walk_status") not in {"skipped", "failed", "blocked"}:
+            continue
+        reason = (
+            row.get("reason_code")
+            or (row.get("raw_payload") or {}).get("walk_gate_reason_code")
+            or "unknown"
+        )
+        counts[str(reason)] += 1
+    return [
+        {"reason_code": reason, "count": count}
+        for reason, count in counts.most_common()
+    ]
+
+
+def _strategy_review_summary(strategy_review: dict[str, Any]) -> dict[str, Any]:
+    v4_summary = strategy_review.get("v4_summary") or {}
+    if not v4_summary and isinstance(strategy_review.get("raw_payload"), dict):
+        v4_summary = strategy_review["raw_payload"].get("v4_summary") or {}
+    return {
+        "review_status": strategy_review.get("review_status", "not_generated"),
+        "v4_summary_schema_version": v4_summary.get("schema_version"),
+        "score_buckets": v4_summary.get("score_buckets", {}),
+        "allow_walk_distribution": v4_summary.get("allow_walk_distribution", {}),
+        "walk_gate_review": v4_summary.get("walk_gate_review", {}),
+    }
+
+
+def _int_value(value: Any) -> int:
+    if isinstance(value, bool):
+        return 0
+    if isinstance(value, int):
+        return value
+    if isinstance(value, float):
+        return int(value)
+    try:
+        return int(str(value))
+    except (TypeError, ValueError):
+        return 0

content_agent/business_modules/result_source_lookup.py

@@ -92,8 +92,11 @@ def run(
                 "decision_action": decision["decision_action"],
                 "decision_reason_code": decision["decision_reason_code"],
                 "search_query_effect_status": decision["search_query_effect_status"],
+                "score": decision.get("score"),
+                "scorecard": decision.get("scorecard", {}),
                 "decision_replay_data": decision.get("decision_replay_data", {}),
                 "source_evidence": decision["source_evidence"],
+                **_v4_explanation_field(decision),
             }
             for decision in decisions
         ],
@@ -157,25 +160,28 @@ def _build_content_assets(
             continue
         path_ids = paths_by_content_id[platform_content_id]
         content_assets.append(
-            {
-                "platform": item["platform"],
-                "platform_content_id": platform_content_id,
-                "policy_run_id": policy_run_id,
-                "content_discovery_id": item["content_discovery_id"],
-                "final_asset_status": "pooled",
-                "decision_id": decision["decision_id"],
-                "rule_pack_id": decision["rule_pack_id"],
-                "rule_pack_version": decision["rule_pack_version"],
-                "strategy_version": decision["strategy_version"],
-                "source_path_record_ids": path_ids,
-                "source_evidence": {
-                    **decision["source_evidence"],
+            _with_v4_explanation(
+                {
+                    "platform": item["platform"],
+                    "platform_content_id": platform_content_id,
+                    "policy_run_id": policy_run_id,
+                    "content_discovery_id": item["content_discovery_id"],
+                    "final_asset_status": "pooled",
+                    "decision_id": decision["decision_id"],
+                    "rule_pack_id": decision["rule_pack_id"],
+                    "rule_pack_version": decision["rule_pack_version"],
+                    "strategy_version": decision["strategy_version"],
                     "source_path_record_ids": path_ids,
+                    "source_evidence": {
+                        **decision["source_evidence"],
+                        "source_path_record_ids": path_ids,
+                    },
+                    "content_media_status": media_by_platform_content_id[
+                        platform_content_id
+                    ]["content_media_status"],
                 },
-                "content_media_status": media_by_platform_content_id[
-                    platform_content_id
-                ]["content_media_status"],
-            }
+                decision,
+            )
         )
     return content_assets
 
@@ -195,27 +201,30 @@ def _build_review_records(
             continue
         path_ids = paths_by_content_id[platform_content_id]
         review_records.append(
-            {
-                "platform": item["platform"],
-                "platform_content_id": platform_content_id,
-                "policy_run_id": policy_run_id,
-                "content_discovery_id": item["content_discovery_id"],
-                "review_status": "pending_review",
-                "final_asset_status": "review_only",
-                "decision_id": decision["decision_id"],
-                "rule_pack_id": decision["rule_pack_id"],
-                "rule_pack_version": decision["rule_pack_version"],
-                "strategy_version": decision["strategy_version"],
-                "decision_reason_code": decision["decision_reason_code"],
-                "source_path_record_ids": path_ids,
-                "source_evidence": {
-                    **decision["source_evidence"],
+            _with_v4_explanation(
+                {
+                    "platform": item["platform"],
+                    "platform_content_id": platform_content_id,
+                    "policy_run_id": policy_run_id,
+                    "content_discovery_id": item["content_discovery_id"],
+                    "review_status": "pending_review",
+                    "final_asset_status": "review_only",
+                    "decision_id": decision["decision_id"],
+                    "rule_pack_id": decision["rule_pack_id"],
+                    "rule_pack_version": decision["rule_pack_version"],
+                    "strategy_version": decision["strategy_version"],
+                    "decision_reason_code": decision["decision_reason_code"],
                     "source_path_record_ids": path_ids,
+                    "source_evidence": {
+                        **decision["source_evidence"],
+                        "source_path_record_ids": path_ids,
+                    },
+                    "content_media_status": media_by_platform_content_id[
+                        platform_content_id
+                    ]["content_media_status"],
                 },
-                "content_media_status": media_by_platform_content_id[
-                    platform_content_id
-                ]["content_media_status"],
-            }
+                decision,
+            )
         )
     return review_records
 
@@ -232,17 +241,70 @@ def _build_reject_records(
         if decision["decision_action"] != "REJECT_CONTENT":
             continue
         reject_records.append(
-            {
-                "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"],
-                "source_evidence": decision["source_evidence"],
-            }
+            _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"],
+                    "source_evidence": decision["source_evidence"],
+                },
+                decision,
+            )
         )
     return reject_records
 
 
+def _with_v4_explanation(
+    record: dict[str, Any],
+    decision: dict[str, Any],
+) -> dict[str, Any]:
+    explanation = _v4_decision_explanation(decision)
+    if explanation:
+        record["v4_explanation"] = explanation
+    return record
+
+
+def _v4_explanation_field(decision: dict[str, Any]) -> dict[str, Any]:
+    explanation = _v4_decision_explanation(decision)
+    return {"v4_explanation": explanation} if explanation else {}
+
+
+def _v4_decision_explanation(decision: dict[str, Any]) -> dict[str, Any]:
+    scorecard = decision.get("scorecard") or {}
+    if scorecard.get("schema_version") != "v4_scorecard.v1":
+        return {}
+    replay_data = decision.get("decision_replay_data") or {}
+    explanation = {
+        "schema_version": "v4_decision_explanation.v1",
+        "scorecard_schema_version": scorecard.get("schema_version"),
+        "query_relevance_score": scorecard.get("query_relevance_score"),
+        "platform_performance_score": scorecard.get("platform_performance_score"),
+        "score": decision.get("score"),
+        "platform_performance_components": scorecard.get("platform_performance_components", []),
+        "missing_observable_fields": scorecard.get("missing_observable_fields", []),
+        "decision_action": decision.get("decision_action"),
+        "decision_reason_code": decision.get("decision_reason_code"),
+        "search_query_effect_status": decision.get("search_query_effect_status"),
+        "allow_walk": replay_data.get("allow_walk"),
+        "allow_walk_reason": replay_data.get("allow_walk_reason"),
+        "walk_gate_snapshot": replay_data.get("walk_gate_snapshot"),
+    }
+    for optional_field in [
+        "score_missing",
+        "failure_type",
+        "exception_type",
+        "http_status_code",
+        "retry_count",
+        "final_status",
+    ]:
+        if optional_field in scorecard:
+            explanation[optional_field] = scorecard[optional_field]
+        elif optional_field in replay_data:
+            explanation[optional_field] = replay_data[optional_field]
+    return explanation
+
+
 def _build_publish_jobs(
     run_id: str,
     policy_run_id: str,

content_agent/business_modules/run_record/validation.py

@@ -126,6 +126,8 @@ def validate_run(run_id: str, runtime: RuntimeFileStore) -> dict[str, Any]:
     _check_v4_score_contract(data, findings)
     _check_v4_walk_gate_contract(data, findings)
     _check_v4_walk_action_consumption(data, findings)
+    _check_v4_final_output_explanation(data, findings)
+    _check_v4_strategy_review_explanation(data, findings)
     _check_v4_action_thresholds(data, findings)
     _check_v4_gemini_failure_contract(data, findings)
     _check_v4_legacy_field_blocklist(data, findings)
@@ -974,6 +976,174 @@ def _check_v4_walk_action_consumption(data: dict[str, Any], findings: list[dict[
             )
 
 
+def _check_v4_final_output_explanation(
+    data: dict[str, Any],
+    findings: list[dict[str, Any]],
+) -> None:
+    v4_decisions_by_id = {
+        decision.get("decision_id"): decision
+        for decision in data.get("rule_decisions.jsonl", [])
+        if _is_v4_contract_record(decision) and decision.get("decision_id")
+    }
+    if not v4_decisions_by_id:
+        return
+
+    records_by_decision_id = _final_output_records_by_decision_id(
+        data.get("final_output.json", {})
+    )
+    for decision_id, decision in v4_decisions_by_id.items():
+        section_records = records_by_decision_id.get(decision_id) or {}
+        decision_records = section_records.get("decision_records") or []
+        if not decision_records:
+            _fail(
+                findings,
+                "v4_final_output_explanation_missing",
+                f"final_output decision_records missing V4 explanation record for decision {decision_id}",
+            )
+            continue
+        for section, records in section_records.items():
+            for record in records:
+                _check_v4_explanation_record(
+                    decision,
+                    record,
+                    f"final_output.{section}",
+                    findings,
+                )
+
+
+def _final_output_records_by_decision_id(
+    final_output: dict[str, Any],
+) -> dict[str, dict[str, list[dict[str, Any]]]]:
+    by_decision_id: dict[str, dict[str, list[dict[str, Any]]]] = {}
+    for section in [
+        "content_assets",
+        "review_records",
+        "reject_records",
+        "decision_records",
+    ]:
+        for record in final_output.get(section, []) or []:
+            decision_id = record.get("decision_id")
+            if not decision_id:
+                for candidate in record.get("decision_ids") or []:
+                    by_decision_id.setdefault(candidate, {}).setdefault(section, []).append(record)
+                continue
+            by_decision_id.setdefault(decision_id, {}).setdefault(section, []).append(record)
+    return by_decision_id
+
+
+def _check_v4_explanation_record(
+    decision: dict[str, Any],
+    record: dict[str, Any],
+    section: str,
+    findings: list[dict[str, Any]],
+) -> None:
+    explanation = record.get("v4_explanation")
+    decision_id = decision.get("decision_id")
+    if not isinstance(explanation, dict) or not explanation:
+        _fail(
+            findings,
+            "v4_final_output_explanation_missing",
+            f"{section} record for decision {decision_id} missing v4_explanation",
+        )
+        return
+    if explanation.get("schema_version") != "v4_decision_explanation.v1":
+        _fail(
+            findings,
+            "v4_final_output_explanation_mismatch",
+            f"{section} record for decision {decision_id} has invalid V4 explanation schema",
+        )
+    scorecard = decision.get("scorecard") or {}
+    replay_data = decision.get("decision_replay_data") or {}
+    expected_values = {
+        "scorecard_schema_version": scorecard.get("schema_version"),
+        "query_relevance_score": scorecard.get("query_relevance_score"),
+        "platform_performance_score": scorecard.get("platform_performance_score"),
+        "score": decision.get("score"),
+        "decision_reason_code": decision.get("decision_reason_code"),
+        "allow_walk": replay_data.get("allow_walk"),
+        "allow_walk_reason": replay_data.get("allow_walk_reason"),
+        "walk_gate_snapshot": replay_data.get("walk_gate_snapshot"),
+    }
+    for field_name, expected in expected_values.items():
+        if expected is None and field_name not in explanation:
+            continue
+        if not _v4_values_equal(explanation.get(field_name), expected):
+            _fail(
+                findings,
+                "v4_final_output_explanation_mismatch",
+                f"{section} record for decision {decision_id} has mismatched {field_name}",
+            )
+    if not isinstance(explanation.get("missing_observable_fields"), list):
+        _fail(
+            findings,
+            "v4_final_output_explanation_mismatch",
+            f"{section} record for decision {decision_id} missing_observable_fields must be a list",
+        )
+
+
+def _check_v4_strategy_review_explanation(
+    data: dict[str, Any],
+    findings: list[dict[str, Any]],
+) -> None:
+    if not any(_is_v4_contract_record(decision) for decision in data.get("rule_decisions.jsonl", [])):
+        return
+    strategy_review = data.get("strategy_review.json") or {}
+    if not strategy_review:
+        return
+    v4_summary = strategy_review.get("v4_summary")
+    if not isinstance(v4_summary, dict):
+        _fail(
+            findings,
+            "v4_strategy_review_explanation_missing",
+            "strategy_review missing v4_summary for V4 decisions",
+        )
+        return
+    required_fields = [
+        "schema_version",
+        "score_buckets",
+        "allow_walk_distribution",
+        "walk_gate_review",
+    ]
+    missing = [field for field in required_fields if field not in v4_summary]
+    if missing:
+        _fail(
+            findings,
+            "v4_strategy_review_explanation_missing",
+            f"strategy_review.v4_summary missing fields: {missing}",
+        )
+        return
+    if v4_summary.get("schema_version") != "v4_strategy_review_summary.v1":
+        _fail(
+            findings,
+            "v4_strategy_review_explanation_invalid",
+            "strategy_review.v4_summary has invalid schema_version",
+        )
+    if not isinstance(v4_summary.get("score_buckets"), dict):
+        _fail(
+            findings,
+            "v4_strategy_review_explanation_invalid",
+            "strategy_review.v4_summary.score_buckets must be an object",
+        )
+    allow_walk_distribution = v4_summary.get("allow_walk_distribution")
+    if not isinstance(allow_walk_distribution, dict) or not {
+        "allowed",
+        "denied",
+        "missing",
+    } <= set(allow_walk_distribution):
+        _fail(
+            findings,
+            "v4_strategy_review_explanation_invalid",
+            "strategy_review.v4_summary.allow_walk_distribution is incomplete",
+        )
+    walk_gate_review = v4_summary.get("walk_gate_review")
+    if not isinstance(walk_gate_review, dict) or "v4_gate_distribution" not in walk_gate_review:
+        _fail(
+            findings,
+            "v4_strategy_review_explanation_invalid",
+            "strategy_review.v4_summary.walk_gate_review is incomplete",
+        )
+
+
 def _check_v4_action_thresholds(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):
@@ -1089,6 +1259,12 @@ def _is_number(value: Any) -> bool:
     return isinstance(value, (int, float)) and not isinstance(value, bool)
 
 
+def _v4_values_equal(left: Any, right: Any) -> bool:
+    if _is_number(left) and _is_number(right):
+        return abs(float(left) - float(right)) <= 0.01
+    return left == right
+
+
 def _result(run_id: str, findings: list[dict[str, Any]]) -> dict[str, Any]:
     return {
         "run_id": run_id,

content_agent/dashboard_service.py

@@ -733,6 +733,8 @@ def _rule_application_summary(
             # content_effect_status 仅旧数据回退(decision 记录从无该字段时原代码恒读 None)。
             "content_effect_status": decision.get("search_query_effect_status")
             or decision.get("content_effect_status"),
+            "scorecard_schema_version": scorecard.get("schema_version"),
+            "v4_explanation": _dashboard_v4_explanation(decision),
             "primary_reason": _reason_label(decision.get("decision_reason_code")),
             "technical_ref": {
                 "decision_id": decision.get("decision_id"),
@@ -819,6 +821,7 @@ def _walk_graph(
             "executed_rule_pack_id": execution.get("executed_rule_pack_id"),
             "budget_tier": action.get("budget_tier"),
             "reason_code": action.get("reason_code"),
+            "v4_gate": _walk_edge_v4_gate_context(action),
         })
     return {
         "nodes": list(nodes.values()),
@@ -827,6 +830,46 @@ def _walk_graph(
     }
 
 
+def _dashboard_v4_explanation(decision: dict[str, Any]) -> dict[str, Any]:
+    existing = decision.get("v4_explanation")
+    if isinstance(existing, dict) and existing:
+        return existing
+    scorecard = decision.get("scorecard") or {}
+    if scorecard.get("schema_version") != "v4_scorecard.v1":
+        return {}
+    replay_data = decision.get("decision_replay_data") or {}
+    return {
+        "schema_version": "v4_decision_explanation.v1",
+        "scorecard_schema_version": scorecard.get("schema_version"),
+        "query_relevance_score": scorecard.get("query_relevance_score"),
+        "platform_performance_score": scorecard.get("platform_performance_score"),
+        "score": decision.get("score"),
+        "platform_performance_components": scorecard.get("platform_performance_components", []),
+        "missing_observable_fields": scorecard.get("missing_observable_fields", []),
+        "decision_action": decision.get("decision_action"),
+        "decision_reason_code": decision.get("decision_reason_code"),
+        "search_query_effect_status": decision.get("search_query_effect_status")
+        or decision.get("content_effect_status"),
+        "allow_walk": replay_data.get("allow_walk"),
+        "allow_walk_reason": replay_data.get("allow_walk_reason"),
+        "walk_gate_snapshot": replay_data.get("walk_gate_snapshot"),
+    }
+
+
+def _walk_edge_v4_gate_context(action: dict[str, Any]) -> dict[str, Any]:
+    raw_payload = action.get("raw_payload") or {}
+    fields = [
+        "decision_id",
+        "allow_walk",
+        "allow_walk_reason",
+        "walk_gate_snapshot",
+        "walk_gate_status",
+        "walk_gate_reason_code",
+    ]
+    context = {field: raw_payload.get(field) for field in fields if field in raw_payload}
+    return context
+
+
 def _primary_failure_reason(
     run_item: dict[str, Any],
     run_events: list[dict[str, Any]],

content_agent/integrations/walk_graph_json.py

@@ -6,6 +6,7 @@ walk_policy 的拍板值可能带 {value, provenance, tbd} 包裹(留痕),load_p
 
 from __future__ import annotations
 
+import os
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Any
@@ -77,6 +78,7 @@ class WalkGraphStore:
 def _unwrap_policy(raw: dict[str, Any]) -> dict[str, Any]:
     policy = dict(raw)
     policy["global"] = {key: _unwrap(value) for key, value in raw["global"].items()}
+    _apply_global_env_overrides(policy["global"])
     policy["edge_permissions"] = {
         action: {edge: _unwrap(perm) for edge, perm in row.items() if edge not in _PERMISSION_META_KEYS}
         for action, row in raw["edge_permissions"].items()
@@ -87,6 +89,27 @@ def _unwrap_policy(raw: dict[str, Any]) -> dict[str, Any]:
     return policy
 
 
+def _apply_global_env_overrides(global_policy: dict[str, Any]) -> None:
+    overrides = {
+        "CONTENT_AGENT_WALK_MAX_TOTAL_ACTIONS_PER_RUN": "max_total_actions_per_run",
+        "CONTENT_AGENT_GEMINI_MAX_WORKERS": "gemini_max_workers",
+    }
+    for env_key, policy_key in overrides.items():
+        value = _optional_positive_int(os.environ.get(env_key))
+        if value is not None:
+            global_policy[policy_key] = value
+
+
+def _optional_positive_int(value: str | None) -> int | None:
+    if value is None or value == "":
+        return None
+    try:
+        parsed = int(value)
+    except ValueError:
+        return None
+    return parsed if parsed > 0 else None
+
+
 def _validate_graph(graph: dict[str, Any], catalog_ids: set[str]) -> list[dict[str, Any]]:
     findings: list[dict[str, Any]] = []
     node_types = {node["node_type"] for node in graph.get("nodes", [])}

content_agent/run_service.py

@@ -616,12 +616,25 @@ def _gemini_video_client_from_env(env: dict[str, str]) -> GeminiVideoClient:
 
 
 def _gemini_calls_cap() -> int | None:
+    override = _optional_positive_int(os.environ.get("CONTENT_AGENT_GEMINI_CALLS_PER_RUN_CAP"))
+    if override is not None:
+        return override
     try:
         return WalkGraphStore().load_policy()["global"]["gemini_calls_per_run_cap"]
     except Exception:
         return None
 
 
+def _optional_positive_int(value: str | None) -> int | None:
+    if value is None or value == "":
+        return None
+    try:
+        parsed = int(value)
+    except ValueError:
+        return None
+    return parsed if parsed > 0 else None
+
+
 class _DeterministicGeminiVideoClient:
     """mock/默认判定 client:固定返回 V4 高相关结果,供本地/smoke 无网跑通。"""
 

content_agent/schemas.py

@@ -161,8 +161,10 @@ class PlatformDescriptor(BaseModel):
     platform: str
     label: str
     status: str | None = None
-    # 该平台真实可得的互动指标(= profile.heat.signals 的 field,按重要度排序;统一键名)
+    # V3 历史展示字段:来自 profile.heat.signals,保留兼容,不等同于 V4 平台表现分。
     heat_fields: list[str] = []
+    # V4 可观测字段:来自 profile.observable_fields,用于解释平台可观测表现。
+    observable_fields: list[str] = []
 
 
 class PlatformCatalogResponse(BaseModel):

scripts/run_v4_m6_real_acceptance.py

@@ -0,0 +1,213 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import json
+import os
+import sys
+from pathlib import Path
+from typing import Any
+
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from content_agent.business_modules.m6_acceptance_report import build_report
+from content_agent.dashboard_service import DashboardService
+from content_agent.run_service import RunService
+from content_agent.schemas import RunStartRequest
+
+
+PLATFORMS = ("douyin", "kuaishou", "shipinhao")
+ALLOWED_DATA_ORIGINS = {"production_db", "mixed_with_runtime_export"}
+SMOKE_DEFAULTS = {
+    "CONTENT_AGENT_GEMINI_CALLS_PER_RUN_CAP": "3",
+    "CONTENT_AGENT_GEMINI_MAX_WORKERS": "3",
+    "CONTENT_AGENT_WALK_MAX_TOTAL_ACTIONS_PER_RUN": "4",
+}
+
+
+def main() -> int:
+    if not _db_runtime_enabled():
+        raise SystemExit("CONTENT_AGENT_DB_RUNTIME_ENABLED=1 is required for M6 real acceptance")
+    _apply_smoke_defaults()
+
+    service = RunService.from_env()
+    dashboard_service = DashboardService.from_runtime(service.runtime)
+    results = []
+
+    for platform in PLATFORMS:
+        state = service.start_run(
+            RunStartRequest(platform=platform, platform_mode="real", strategy_version="V4")
+        )
+        run_id = state["run_id"]
+        result = _collect_run_result(service, dashboard_service, state, platform)
+        results.append(result)
+        result["acceptance_status"] = _acceptance_status(result)
+
+    payload = {
+        "schema_version": "v4_m6_real_acceptance.v1",
+        "source_kind": "db_default_safe_source",
+        "smoke_limits": _smoke_limits(),
+        "platforms": list(PLATFORMS),
+        "results": results,
+        "status": "pass" if all(row["acceptance_status"] == "pass" for row in results) else "fail",
+    }
+    print(json.dumps(payload, ensure_ascii=False, indent=2, default=str))
+    return 0 if payload["status"] == "pass" else 1
+
+
+def _apply_smoke_defaults() -> None:
+    for key, value in SMOKE_DEFAULTS.items():
+        os.environ.setdefault(key, value)
+
+
+def _smoke_limits() -> dict[str, str | None]:
+    return {key: os.environ.get(key) for key in SMOKE_DEFAULTS}
+
+
+def _collect_run_result(
+    service: RunService,
+    dashboard_service: DashboardService,
+    state: dict[str, Any],
+    platform: str,
+) -> dict[str, Any]:
+    run_id = state["run_id"]
+    summary = _safe_call(lambda: service.get_summary(run_id), {})
+    validation = _safe_call(lambda: service.validate_run(run_id), {"status": "fail", "findings": []})
+    dashboard = _safe_call(lambda: dashboard_service.dashboard(run_id), {})
+    timeline = _safe_call(lambda: dashboard_service.timeline(run_id), {})
+    strategy_review = _safe_call(lambda: service.strategy_review(run_id), {})
+    runtime_files = _safe_call(lambda: dashboard_service.runtime_files(run_id), {})
+    report = _safe_call(
+        lambda: build_report(run_id, service.runtime, platform=platform, platform_mode="real"),
+        {},
+    )
+    data_origin = dashboard.get("data_origin") or runtime_files.get("data_origin")
+
+    return {
+        "platform": platform,
+        "platform_mode": "real",
+        "run_id": run_id,
+        "policy_run_id": state.get("policy_run_id") or summary.get("policy_run_id"),
+        "status": state.get("status"),
+        "validation_status": validation.get("status"),
+        "output_dir": summary.get("output_dir"),
+        "data_origin": data_origin,
+        "db_run_record_present": _db_run_record_present(dashboard, data_origin),
+        "runtime_files": _runtime_file_status(runtime_files),
+        "real_interface_progress": _real_interface_progress(runtime_files, timeline),
+        "dashboard_summary": dashboard.get("summary", {}),
+        "timeline_summary": timeline.get("summary", {}),
+        "strategy_review_status": strategy_review.get("review_status"),
+        "failure_classification": _failure_classification(state, dashboard),
+        "m6_report": report,
+    }
+
+
+def _safe_call(fn: Any, fallback: Any) -> Any:
+    try:
+        return fn()
+    except Exception as exc:
+        return {"error": type(exc).__name__, "message": str(exc)} if isinstance(fallback, dict) else fallback
+
+
+def _runtime_file_status(runtime_files: dict[str, Any]) -> dict[str, bool]:
+    return {
+        item.get("filename"): bool(item.get("exists"))
+        for item in runtime_files.get("files", [])
+        if item.get("filename")
+    }
+
+
+def _real_interface_progress(
+    runtime_files: dict[str, Any],
+    timeline: dict[str, Any],
+) -> dict[str, bool]:
+    file_status = _runtime_file_status(runtime_files)
+    stages = (timeline.get("summary") or {}).get("stage_duration_ms") or {}
+    return {
+        "source_loaded": bool(file_status.get("source_context.json")),
+        "query_generated": bool(file_status.get("search_queries.jsonl")),
+        "platform_attempted": "search_platform" in stages
+        or bool(file_status.get("discovered_content_items.jsonl")),
+        "scoring_attempted": "evaluate_rules" in stages
+        or bool(file_status.get("rule_decisions.jsonl")),
+        "final_output_generated": bool(file_status.get("final_output.json")),
+    }
+
+
+def _acceptance_status(result: dict[str, Any]) -> str:
+    if not result["db_run_record_present"]:
+        return "fail"
+    if not result["real_interface_progress"]["query_generated"]:
+        return "blocked_before_platform"
+    if result["status"] == "failed" and not result["failure_classification"]:
+        return "fail"
+    return "pass"
+
+
+def _db_run_record_present(dashboard: dict[str, Any], data_origin: str | None) -> bool:
+    if data_origin not in ALLOWED_DATA_ORIGINS:
+        return False
+    summary = dashboard.get("summary") or {}
+    return bool(summary.get("run_id"))
+
+
+def _failure_classification(
+    state: dict[str, Any],
+    dashboard: dict[str, Any],
+) -> dict[str, Any] | None:
+    if state.get("status") not in {"failed", "partial_success"}:
+        return None
+    primary = dashboard.get("primary_failure_reason") or {}
+    error_code = state.get("error_code") or primary.get("reason_code")
+    error_detail = state.get("error_detail") or {}
+    return {
+        "category": _failure_category(error_code, error_detail),
+        "error_code": error_code,
+        "message": state.get("error_message") or primary.get("message"),
+        "error_detail": error_detail,
+        "primary_failure_reason": primary,
+    }
+
+
+def _failure_category(error_code: str | None, detail: dict[str, Any]) -> str:
+    code = str(error_code or "")
+    text = json.dumps(detail, ensure_ascii=False, default=str).lower()
+    if code in {"PLATFORM_CONFIG_MISSING", "DB_CONFIG_MISSING", "INVALID_SOURCE", "INVALID_REQUEST"}:
+        return "configuration"
+    if code in {"PLATFORM_RATE_LIMITED"} or "429" in text or "rate" in text:
+        return "rate_limit"
+    if code.startswith("PLATFORM_"):
+        return "upstream_platform"
+    if "gemini" in text or "openrouter" in text:
+        return "gemini_technical_failure"
+    if "timeout" in text or "network" in text or "connection" in text:
+        return "network_or_vpn"
+    return "runtime_or_validator"
+
+
+def _db_runtime_enabled() -> bool:
+    value = os.environ.get("CONTENT_AGENT_DB_RUNTIME_ENABLED")
+    if value is None:
+        value = _env_file_value("CONTENT_AGENT_DB_RUNTIME_ENABLED")
+    return str(value or "").lower() in {"1", "true", "yes", "on"}
+
+
+def _env_file_value(key: str) -> str | None:
+    try:
+        lines = open(".env", encoding="utf-8").read().splitlines()
+    except FileNotFoundError:
+        return None
+    for line in lines:
+        stripped = line.strip()
+        if not stripped or stripped.startswith("#") or "=" not in stripped:
+            continue
+        name, value = stripped.split("=", 1)
+        if name.strip() == key:
+            return value.strip().strip('"').strip("'")
+    return None
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/summarize_v4_m6_run.py

@@ -0,0 +1,40 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from content_agent.business_modules.m6_acceptance_report import build_report
+from content_agent.integrations.runtime_files import LocalRuntimeFileStore
+
+
+def main() -> int:
+    args = _parse_args()
+    runtime = LocalRuntimeFileStore(args.runtime_root)
+    report = build_report(
+        args.run_id,
+        runtime,
+        platform=args.platform,
+        platform_mode=args.platform_mode,
+    )
+    print(json.dumps(report, ensure_ascii=False, indent=2, default=str))
+    return 0
+
+
+def _parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Summarize a V4 M6 run from runtime files.")
+    parser.add_argument("--run-id", required=True)
+    parser.add_argument("--runtime-root", default=Path("runtime/v1"), type=Path)
+    parser.add_argument("--platform", default=None)
+    parser.add_argument("--platform-mode", default="real")
+    return parser.parse_args()
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tech_documents/工程落地/12——V4阶段开发计划.md

@@ -1027,3 +1027,17 @@ DB/runtime:
 - `50+ 粉丝占比 >60% 且 TGI >120` 是否需要调高或调低。
 - Query/平台表现阈值是否按真实入池质量微调。
 - 是否把某些高价值 JSON 字段晋升为正式 DB 列。
+
+## 19. Debug 记录:V4 M6 真实 E2E
+
+2026-06-15 在阿里云海外开发机 `/home/sam/ContentFindAgentNew` 执行 M6 三平台真实 E2E。远端 `.env` 使用 `CONTENT_SUPPLY_DB_HOST=127.0.0.1`、`CONTENT_AGENT_DB_RUNTIME_ENABLED=1`;DB 只读校验 `found_table_count=21`、`schema_ready=true`。
+
+修复点:
+- M6 首轮真实 run 证明平台/Gemini/DB 可通,但默认策略会把 smoke 扩大成较长 walk。已补 `CONTENT_AGENT_GEMINI_CALLS_PER_RUN_CAP`、`CONTENT_AGENT_GEMINI_MAX_WORKERS`、`CONTENT_AGENT_WALK_MAX_TOTAL_ACTIONS_PER_RUN` 环境变量覆盖,只在显式 smoke 环境收窄规模,不改 JSON/Excel/config 和 M3/M4/M5 业务语义。
+- 调试残留 `v1_run_265b1d2e07da` 已标记为 `M6_DEBUG_ABORTED`,不计入验收。
+
+最终验收:
+- `douyin`: `v1_run_815c71aee224`, `success`, `validation_status=pass`, `data_origin=production_db`, 总耗时 274435ms。
+- `kuaishou`: `v1_run_24f25c36eb9f`, `success`, `validation_status=pass`, `data_origin=production_db`, 总耗时 57780ms。
+- `shipinhao`: `v1_run_3d285f2f5f8f`, `success`, `validation_status=pass`, `data_origin=production_db`, 总耗时 76383ms。
+- 总验收报告 `status=pass`;三平台均真实生成 `search_queries/discovered_content_items/content_media_records/pattern_recall_evidence/rule_decisions/walk_actions/final_output/strategy_review`,并通过 DB runtime 回读、dashboard summary、timeline、final_output、strategy_review 与 `validate_run()`。

tests/test_api.py

@@ -78,7 +78,18 @@ def test_api_runs_and_queries_mock_chain(tmp_path, monkeypatch):
     source_stage = next(stage for stage in dashboard["stage_conclusions"] if stage["stage_id"] == "source")
     assert source_stage["detail"] == "需求池 ID：1"
     assert isinstance(dashboard["rule_application_summary"], list)
+    v4_rule_rows = [
+        row
+        for row in dashboard["rule_application_summary"]
+        if row.get("scorecard_schema_version") == "v4_scorecard.v1"
+    ]
+    assert v4_rule_rows
+    assert all(row["v4_explanation"]["scorecard_schema_version"] == "v4_scorecard.v1" for row in v4_rule_rows)
     assert "nodes" in dashboard["walk_graph"]
+    v4_gate_edges = [
+        edge for edge in dashboard["walk_graph"]["edges"] if edge.get("v4_gate")
+    ]
+    assert all(isinstance(edge["v4_gate"], dict) for edge in v4_gate_edges)
     assert dashboard["technical_refs"]["runtime_files_url"].endswith("/runtime-files")
 
     queries = client.get(f"/runs/{run_id}/queries").json()

tests/test_database_runtime.py

@@ -762,6 +762,70 @@ def test_database_runtime_update_final_output_upserts_validation_status():
     assert json.loads(values["summary"])["run_path_complete"] is True
 
 
+def test_database_runtime_preserves_m5_explanation_payloads():
+    connection = FakeConnection()
+    store = DatabaseRuntimeStore(_config(), connection_factory=lambda: connection)
+    explanation = {
+        "schema_version": "v4_decision_explanation.v1",
+        "query_relevance_score": 80,
+        "platform_performance_score": 70,
+        "score": 75,
+        "allow_walk": True,
+        "walk_gate_snapshot": {"score": 75},
+    }
+    v4_summary = {
+        "schema_version": "v4_strategy_review_summary.v1",
+        "score_buckets": {"pool": 1},
+        "allow_walk_distribution": {"allowed": 1, "denied": 0, "missing": 0},
+        "walk_gate_review": {"v4_gate_distribution": {"allowed": 1}},
+    }
+
+    store.update_json(
+        "run_001",
+        "final_output.json",
+        {
+            "schema_version": "runtime_record.v1",
+            "run_id": "run_001",
+            "policy_run_id": "policy_run_001",
+            "summary": {},
+            "validation_status": "pass",
+            "decision_records": [
+                {
+                    "decision_id": "decision_001",
+                    "v4_explanation": explanation,
+                }
+            ],
+        },
+    )
+    store.update_json(
+        "run_001",
+        "strategy_review.json",
+        {
+            "schema_version": "runtime_record.v1",
+            "run_id": "run_001",
+            "policy_run_id": "policy_run_001",
+            "review_id": "review_001",
+            "review_status": "generated",
+            "summary": {},
+            "v4_summary": v4_summary,
+            "effective_search_queries": [],
+            "weak_search_queries": [],
+            "top_reject_reasons": [],
+            "productive_paths": [],
+            "suggestions": [],
+            "raw_payload": {"v4_summary": v4_summary},
+        },
+    )
+
+    final_values = _insert_values(*connection.statements[-2])
+    final_payload = json.loads(final_values["final_output"])
+    assert final_payload["decision_records"][0]["v4_explanation"] == explanation
+
+    review_values = _insert_values(*connection.statements[-1])
+    review_payload = json.loads(review_values["raw_payload"])
+    assert review_payload["v4_summary"] == v4_summary
+
+
 def test_database_runtime_reads_performance_feedback_payloads():
     connection = FakeConnection()
     connection.select_all_result = [

tests/test_p7_final_output.py

@@ -48,3 +48,37 @@ def test_run_service_rewrites_final_output_with_final_validation_status(tmp_path
     assert final_output["validation_status"] == validation["status"]
     assert final_output["summary"]["run_path_complete"] is True
     assert final_output["summary"]["trace_complete"] is True
+
+
+def test_final_output_carries_v4_explanation_records(tmp_path):
+    service, run_id = _start_mock_run(tmp_path)
+
+    final_output = service.read_json(run_id, "final_output.json")
+    v4_decision_records = [
+        record
+        for record in final_output["decision_records"]
+        if (record.get("scorecard") or {}).get("schema_version") == "v4_scorecard.v1"
+    ]
+
+    assert v4_decision_records
+    for record in v4_decision_records:
+        explanation = record["v4_explanation"]
+        assert explanation["schema_version"] == "v4_decision_explanation.v1"
+        assert explanation["scorecard_schema_version"] == "v4_scorecard.v1"
+        assert explanation["query_relevance_score"] == record["scorecard"]["query_relevance_score"]
+        assert explanation["platform_performance_score"] == record["scorecard"]["platform_performance_score"]
+        assert explanation["score"] == record["score"]
+        assert "allow_walk" in explanation
+        assert "walk_gate_snapshot" in explanation
+
+    v4_ids = {record["decision_id"] for record in v4_decision_records}
+    section_records = (
+        final_output["content_assets"]
+        + final_output["review_records"]
+        + final_output["reject_records"]
+    )
+    assert [
+        record
+        for record in section_records
+        if record.get("decision_id") in v4_ids and record.get("v4_explanation")
+    ]

tests/test_p8_strategy_review.py

@@ -83,6 +83,47 @@ def test_strategy_review_filters_legacy_contract_values(tmp_path):
     assert "HOLD_CONTENT_" + "PENDING" not in review["rule_review"]["decision_distribution"]
 
 
+def test_strategy_review_builds_v4_single_run_summary(tmp_path):
+    runtime = LocalRuntimeFileStore(tmp_path / "runtime")
+    run_id = "run_v4"
+    policy_run_id = "policy_v4"
+    runtime.prepare_run(run_id)
+    _write_v4_runtime(runtime, run_id, policy_run_id)
+
+    review = learning_review.run(run_id, policy_run_id, runtime)
+
+    v4_summary = review["v4_summary"]
+    assert v4_summary["schema_version"] == "v4_strategy_review_summary.v1"
+    assert v4_summary["summary_status"] == "available"
+    assert v4_summary["v4_decision_count"] == 4
+    assert v4_summary["score_buckets"] == {
+        "pool": 1,
+        "review": 1,
+        "reject": 1,
+        "technical_retry": 1,
+        "unknown": 0,
+    }
+    assert v4_summary["missing_observable_fields"] == [
+        {"field": "statistics.play_count", "reason": "natural_platform_missing", "count": 1},
+        {"field": "statistics.share_count", "reason": "runtime_missing", "count": 1},
+    ]
+    assert v4_summary["technical_retry"]["count"] == 1
+    assert v4_summary["allow_walk_distribution"] == {
+        "allowed": 1,
+        "denied": 3,
+        "missing": 0,
+    }
+    assert v4_summary["walk_gate_review"]["v4_gate_distribution"]["allowed"] == 1
+    assert v4_summary["walk_gate_review"]["v4_gate_distribution"]["denied"] == 1
+    assert v4_summary["walk_gate_review"]["deny_reasons"] == [
+        {"reason_code": "v4_allow_walk_denied", "count": 1}
+    ]
+    assert review["walk_review"]["deny_reasons"] == [
+        {"reason_code": "v4_allow_walk_denied", "count": 1}
+    ]
+    assert runtime.read_json(run_id, "strategy_review.json")["raw_payload"]["v4_summary"]
+
+
 def _write_minimal_runtime(runtime, run_id: str, policy_run_id: str) -> None:
     runtime.write_json(
         run_id,
@@ -234,3 +275,140 @@ def _decision(
         "source_evidence": {"search_query_id": search_query_id} if search_query_id else {},
         "raw_payload": {},
     }
+
+
+def _write_v4_runtime(runtime, run_id: str, policy_run_id: str) -> None:
+    runtime.write_json(
+        run_id,
+        "final_output.json",
+        {
+            "schema_version": "runtime_record.v1",
+            "run_id": run_id,
+            "policy_run_id": policy_run_id,
+            "policy": {"strategy_version": "V4"},
+            "walk_strategy": {"walk_strategy_version": "V4"},
+            "content_assets": [{"content_asset_id": "asset_001"}],
+            "author_assets": [],
+            "summary": {
+                "search_query_count": 4,
+                "discovered_content_count": 4,
+                "pooled_content_count": 1,
+                "review_content_count": 2,
+                "rejected_content_count": 1,
+            },
+        },
+    )
+    runtime.append_jsonl(
+        run_id,
+        "search_clues.jsonl",
+        [
+            _clue(run_id, policy_run_id, "clue_v4", "q_v4", "父爱感悟", "success", 1)
+        ],
+    )
+    runtime.append_jsonl(
+        run_id,
+        "rule_decisions.jsonl",
+        [
+            _v4_decision(run_id, policy_run_id, "decision_pool", "content_pool", "ADD_TO_CONTENT_POOL", "success", 80, 70, True, []),
+            _v4_decision(run_id, policy_run_id, "decision_review", "content_review", "KEEP_CONTENT_FOR_REVIEW", "pending", 60, 60, False, [{"field": "statistics.share_count", "reason": "runtime_missing"}]),
+            _v4_decision(run_id, policy_run_id, "decision_reject", "content_reject", "REJECT_CONTENT", "failed", 40, 70, False, [{"field": "statistics.play_count", "reason": "natural_platform_missing"}]),
+            _v4_decision(run_id, policy_run_id, "decision_retry", "content_retry", "KEEP_CONTENT_FOR_REVIEW", "pending", None, None, False, [], reason="v4_technical_retry_needed"),
+        ],
+    )
+    runtime.append_jsonl(run_id, "discovered_content_items.jsonl", [])
+    runtime.append_jsonl(run_id, "source_path_records.jsonl", [])
+    runtime.append_jsonl(
+        run_id,
+        "walk_actions.jsonl",
+        [
+            {
+                "run_id": run_id,
+                "policy_run_id": policy_run_id,
+                "walk_action_id": "walk_allowed",
+                "edge_id": "hashtag_to_query",
+                "walk_status": "success",
+                "raw_payload": {
+                    "decision_id": "decision_pool",
+                    "allow_walk": True,
+                    "allow_walk_reason": "query>=70/platform>=65/score>=70",
+                    "walk_gate_status": "allowed",
+                    "walk_gate_snapshot": {"query_relevance_score": 80, "platform_performance_score": 70, "score": 75},
+                },
+            },
+            {
+                "run_id": run_id,
+                "policy_run_id": policy_run_id,
+                "walk_action_id": "walk_denied",
+                "edge_id": "author_to_works",
+                "walk_status": "skipped",
+                "reason_code": "v4_allow_walk_denied",
+                "raw_payload": {
+                    "decision_id": "decision_review",
+                    "allow_walk": False,
+                    "allow_walk_reason": "score below allow_walk threshold",
+                    "walk_gate_status": "denied",
+                    "walk_gate_reason_code": "v4_allow_walk_denied",
+                    "walk_gate_snapshot": {"query_relevance_score": 60, "platform_performance_score": 60, "score": 60},
+                },
+            },
+        ],
+    )
+
+
+def _v4_decision(
+    run_id,
+    policy_run_id,
+    decision_id,
+    target_id,
+    action,
+    effect_status,
+    query_score,
+    platform_score,
+    allow_walk,
+    missing_fields,
+    reason=None,
+):
+    score = (
+        round(query_score * 0.5 + platform_score * 0.5, 2)
+        if query_score is not None and platform_score is not None
+        else None
+    )
+    return {
+        "run_id": run_id,
+        "policy_run_id": policy_run_id,
+        "decision_id": decision_id,
+        "decision_target_type": "content",
+        "decision_target_id": target_id,
+        "decision_action": action,
+        "decision_reason_code": reason or (
+            "v4_query_and_platform_pass"
+            if action == "ADD_TO_CONTENT_POOL"
+            else "v4_score_review_needed"
+            if action == "KEEP_CONTENT_FOR_REVIEW"
+            else "v4_query_or_score_below_threshold"
+        ),
+        "search_query_effect_status": effect_status,
+        "score": score,
+        "scorecard": {
+            "schema_version": "v4_scorecard.v1",
+            "query_relevance_score": query_score,
+            "platform_performance_score": platform_score,
+            "missing_observable_fields": missing_fields,
+            "score_missing": score is None,
+            "failure_type": "llm_timeout" if score is None else None,
+        },
+        "decision_replay_data": {
+            "allow_walk": allow_walk,
+            "allow_walk_reason": (
+                "query>=70/platform>=65/score>=70"
+                if allow_walk
+                else "score below allow_walk threshold"
+            ),
+            "walk_gate_snapshot": {
+                "query_relevance_score": query_score,
+                "platform_performance_score": platform_score,
+                "score": score,
+            },
+        },
+        "raw_payload": {},
+    }

tests/test_v4_m5_explanation_replay.py

@@ -0,0 +1,97 @@
+import json
+
+from fastapi.testclient import TestClient
+
+from content_agent import api
+from content_agent.business_modules import learning_review, result_source_lookup
+from content_agent.business_modules.run_record.validation import validate_run
+from content_agent.integrations.database_runtime import DatabaseRuntimeStore
+from content_agent.integrations.runtime_files import LocalRuntimeFileStore, RUNTIME_FILENAMES
+from content_agent.run_service import RunService
+from tests.test_database_runtime import FakeConnection, _config, _insert_values
+from tests.test_v4_validator_contract import POLICY_RUN_ID, RUN_ID, _runtime_payload
+
+
+def test_v4_m5_explanation_replay_api_and_db(tmp_path, monkeypatch):
+    runtime = LocalRuntimeFileStore(tmp_path / "runtime")
+    runtime.prepare_run(RUN_ID)
+    payload = _runtime_payload()
+    for filename in RUNTIME_FILENAMES:
+        if filename in {"final_output.json", "strategy_review.json"}:
+            continue
+        value = payload[filename]
+        if isinstance(value, list):
+            runtime.append_jsonl(RUN_ID, filename, value)
+        else:
+            runtime.write_json(RUN_ID, filename, value)
+    content_media_records = [
+        {
+            **record,
+            "content_media_status": record.get("content_media_status")
+            or record.get("media_status")
+            or "metadata_only",
+        }
+        for record in payload["content_media_records.jsonl"]
+    ]
+    policy_bundle = _policy_bundle()
+
+    final_output = result_source_lookup.run(
+        RUN_ID,
+        POLICY_RUN_ID,
+        policy_bundle,
+        payload["discovered_content_items.jsonl"],
+        content_media_records,
+        payload["rule_decisions.jsonl"],
+        payload["source_path_records.jsonl"],
+        payload["search_clues.jsonl"],
+        runtime,
+    )
+    strategy_review = learning_review.run(RUN_ID, POLICY_RUN_ID, runtime)
+    validation = validate_run(RUN_ID, runtime)
+
+    assert validation["status"] == "pass", validation["findings"]
+    assert final_output["decision_records"][0]["v4_explanation"]["allow_walk"] is True
+    assert strategy_review["v4_summary"]["schema_version"] == "v4_strategy_review_summary.v1"
+
+    monkeypatch.setattr(api, "service", RunService(runtime_root=tmp_path / "runtime"))
+    client = TestClient(api.app)
+    dashboard = client.get(f"/runs/{RUN_ID}/dashboard")
+    final_response = client.get(f"/runs/{RUN_ID}/final-output")
+    review_response = client.get(f"/runs/{RUN_ID}/strategy-review")
+
+    assert dashboard.status_code == 200
+    assert final_response.status_code == 200
+    assert review_response.status_code == 200
+    dashboard_row = dashboard.json()["rule_application_summary"][0]
+    assert dashboard_row["v4_explanation"]["scorecard_schema_version"] == "v4_scorecard.v1"
+    assert final_response.json()["data"]["decision_records"][0]["v4_explanation"]
+    assert review_response.json()["data"]["v4_summary"]["allow_walk_distribution"]["allowed"] == 1
+
+    connection = FakeConnection()
+    store = DatabaseRuntimeStore(_config(), connection_factory=lambda: connection)
+    store.update_json(RUN_ID, "final_output.json", final_output)
+    store.update_json(RUN_ID, "strategy_review.json", strategy_review)
+
+    final_values = _insert_values(*connection.statements[-2])
+    review_values = _insert_values(*connection.statements[-1])
+    assert json.loads(final_values["final_output"])["decision_records"][0]["v4_explanation"]
+    assert json.loads(review_values["raw_payload"])["v4_summary"]["score_buckets"]["pool"] == 1
+
+
+def _policy_bundle() -> dict:
+    return {
+        "policy_bundle_id": "policy_bundle_v4",
+        "strategy_id": "strategy_v4",
+        "strategy_version": "V4",
+        "rule_pack_id": "douyin_content_discovery_rule_pack_v4",
+        "rule_pack_version": "4.0.0",
+        "policy_bundle_hash": "hash_v4",
+        "strategy_source_ref": {"path": "strategy"},
+        "rule_pack_source_ref": {"path": "rule_pack"},
+        "walk_strategy_id": "douyin_walk_strategy_v1",
+        "walk_strategy_version": "V4",
+        "walk_strategy_source_ref": {"path": "walk_strategy"},
+        "dispatch": {},
+        "dispatch_id": "dispatch_v4",
+        "runtime_status_contract": {},
+    }

tests/test_v4_m6_real_acceptance.py

@@ -0,0 +1,39 @@
+import os
+
+import pytest
+
+from scripts.run_v4_m6_real_acceptance import ALLOWED_DATA_ORIGINS, main
+
+
+def test_v4_m6_real_acceptance_requires_explicit_opt_in(monkeypatch):
+    monkeypatch.delenv("CONTENT_AGENT_M6_REAL_ACCEPTANCE", raising=False)
+
+    if os.environ.get("CONTENT_AGENT_M6_REAL_ACCEPTANCE") != "1":
+        pytest.skip("M6 real acceptance requires CONTENT_AGENT_M6_REAL_ACCEPTANCE=1")
+
+
+@pytest.mark.skipif(
+    os.environ.get("CONTENT_AGENT_M6_REAL_ACCEPTANCE") != "1",
+    reason="M6 real acceptance must be explicitly enabled",
+)
+def test_v4_m6_real_acceptance_runs_three_real_platforms(capsys):
+    exit_code = main()
+    captured = capsys.readouterr()
+
+    assert exit_code == 0, captured.out
+    import json
+
+    payload = json.loads(captured.out)
+    assert payload["status"] == "pass"
+    assert [row["platform"] for row in payload["results"]] == [
+        "douyin",
+        "kuaishou",
+        "shipinhao",
+    ]
+    for row in payload["results"]:
+        assert row["run_id"].startswith("v1_run_")
+        assert row["db_run_record_present"] is True
+        assert row["data_origin"] in ALLOWED_DATA_ORIGINS
+        assert row["m6_report"]["schema_version"] == "v4_m6_acceptance_report.v1"
+        if row["status"] == "failed":
+            assert row["failure_classification"]

tests/test_v4_validator_contract.py

@@ -40,6 +40,7 @@ def test_v4_contract_does_not_apply_to_v3_scorecard_records(tmp_path):
 def test_v4_score_contract_rejects_bad_total(tmp_path):
     def mutate(data: dict[str, Any]) -> None:
         data["rule_decisions.jsonl"][0]["score"] = 92
+        _refresh_final_output_explanations(data)
 
     runtime = _write_runtime(tmp_path, mutate)
 
@@ -54,6 +55,7 @@ def test_v4_walk_gate_rejects_allow_walk_below_threshold(tmp_path):
         decision["score"] = 65
         decision["scorecard"]["query_relevance_score"] = 60
         decision["scorecard"]["platform_performance_score"] = 70
+        _refresh_final_output_explanations(data)
 
     runtime = _write_runtime(tmp_path, mutate)
 
@@ -65,6 +67,7 @@ def test_v4_walk_gate_rejects_allow_walk_below_threshold(tmp_path):
 def test_v4_walk_gate_rejects_allow_walk_false_at_passing_threshold(tmp_path):
     def mutate(data: dict[str, Any]) -> None:
         data["rule_decisions.jsonl"][0]["decision_replay_data"]["allow_walk"] = False
+        _refresh_final_output_explanations(data)
 
     runtime = _write_runtime(tmp_path, mutate)
 
@@ -85,6 +88,7 @@ def test_v4_walk_action_rejects_success_when_allow_walk_false(tmp_path):
             "platform_performance_score": 60,
             "score": 70,
         }
+        _refresh_final_output_explanations(data)
         data["walk_actions.jsonl"].append(
             _walk_action(
                 "walk_page_001",
@@ -174,6 +178,39 @@ def test_v4_legacy_field_blocklist_rejects_v4_records_only(tmp_path):
     assert _check_ids(result).count("v4_legacy_field_present") == 2
 
 
+def test_v4_final_output_requires_explanation(tmp_path):
+    def mutate(data: dict[str, Any]) -> None:
+        data["final_output.json"]["decision_records"][0].pop("v4_explanation")
+
+    runtime = _write_runtime(tmp_path, mutate)
+
+    result = validate_run(RUN_ID, runtime)
+
+    assert "v4_final_output_explanation_missing" in _check_ids(result)
+
+
+def test_v4_final_output_explanation_must_match_decision(tmp_path):
+    def mutate(data: dict[str, Any]) -> None:
+        data["final_output.json"]["decision_records"][0]["v4_explanation"]["allow_walk"] = False
+
+    runtime = _write_runtime(tmp_path, mutate)
+
+    result = validate_run(RUN_ID, runtime)
+
+    assert "v4_final_output_explanation_mismatch" in _check_ids(result)
+
+
+def test_v4_strategy_review_requires_summary_when_generated(tmp_path):
+    def mutate(data: dict[str, Any]) -> None:
+        data["strategy_review.json"].pop("v4_summary")
+
+    runtime = _write_runtime(tmp_path, mutate)
+
+    result = validate_run(RUN_ID, runtime)
+
+    assert "v4_strategy_review_explanation_missing" in _check_ids(result)
+
+
 def _write_runtime(
     tmp_path,
     mutate: Callable[[dict[str, Any]], None] | None = None,
@@ -247,6 +284,7 @@ def _runtime_payload() -> dict[str, Any]:
             "dispatch_id": "dispatch_v4",
             "strategy_version": "V4",
             "allow_walk": True,
+            "allow_walk_reason": "query>=70/platform>=65/score>=70",
             "walk_gate_snapshot": {
                 "query_relevance_score": 80,
                 "platform_performance_score": 70,
@@ -255,6 +293,7 @@ def _runtime_payload() -> dict[str, Any]:
         },
         "raw_payload": {"decision_id": "decision_001", "v4_contract": True},
     }
+    v4_explanation = _v4_explanation(decision)
     return {
         "source_context.json": {
             "schema_version": "runtime_record.v1",
@@ -380,6 +419,7 @@ def _runtime_payload() -> dict[str, Any]:
                     "decision_id": "decision_001",
                     "source_path_record_ids": path_ids,
                     "source_evidence": copy.deepcopy(source_evidence),
+                    "v4_explanation": copy.deepcopy(v4_explanation),
                 }
             ],
             "author_assets": [],
@@ -387,7 +427,11 @@ def _runtime_payload() -> dict[str, Any]:
             "decision_records": [
                 {
                     "decision_id": "decision_001",
+                    "score": 75,
+                    "scorecard": copy.deepcopy(decision["scorecard"]),
+                    "decision_replay_data": copy.deepcopy(decision["decision_replay_data"]),
                     "source_evidence": copy.deepcopy(source_evidence),
+                    "v4_explanation": copy.deepcopy(v4_explanation),
                 }
             ],
             "search_clues": [],
@@ -406,11 +450,69 @@ def _runtime_payload() -> dict[str, Any]:
             "run_id": RUN_ID,
             "policy_run_id": POLICY_RUN_ID,
             "summary": {},
+            "v4_summary": {
+                "schema_version": "v4_strategy_review_summary.v1",
+                "score_buckets": {
+                    "pool": 1,
+                    "review": 0,
+                    "reject": 0,
+                    "technical_retry": 0,
+                    "unknown": 0,
+                },
+                "allow_walk_distribution": {
+                    "allowed": 1,
+                    "denied": 0,
+                    "missing": 0,
+                },
+                "walk_gate_review": {
+                    "v4_gate_distribution": {
+                        "allowed": 0,
+                        "denied": 0,
+                        "missing": 0,
+                    }
+                },
+            },
             "raw_payload": {"strategy_review_id": "review_001"},
         },
     }
 
 
+def _v4_explanation(decision: dict[str, Any]) -> dict[str, Any]:
+    scorecard = decision["scorecard"]
+    replay = decision["decision_replay_data"]
+    return {
+        "schema_version": "v4_decision_explanation.v1",
+        "scorecard_schema_version": scorecard["schema_version"],
+        "query_relevance_score": scorecard["query_relevance_score"],
+        "platform_performance_score": scorecard["platform_performance_score"],
+        "score": decision["score"],
+        "platform_performance_components": scorecard.get("platform_performance_components", []),
+        "missing_observable_fields": scorecard.get("missing_observable_fields", []),
+        "decision_action": decision["decision_action"],
+        "decision_reason_code": decision["decision_reason_code"],
+        "search_query_effect_status": decision["search_query_effect_status"],
+        "allow_walk": replay["allow_walk"],
+        "allow_walk_reason": replay["allow_walk_reason"],
+        "walk_gate_snapshot": replay["walk_gate_snapshot"],
+    }
+
+
+def _refresh_final_output_explanations(data: dict[str, Any]) -> None:
+    decisions = {decision["decision_id"]: decision for decision in data["rule_decisions.jsonl"]}
+    for section in ["content_assets", "review_records", "reject_records", "decision_records"]:
+        for record in data["final_output.json"].get(section, []):
+            decision = decisions.get(record.get("decision_id"))
+            if not decision:
+                continue
+            record["v4_explanation"] = _v4_explanation(decision)
+            if section == "decision_records":
+                record["score"] = decision.get("score")
+                record["scorecard"] = copy.deepcopy(decision.get("scorecard"))
+                record["decision_replay_data"] = copy.deepcopy(
+                    decision.get("decision_replay_data")
+                )
+
+
 def _source_evidence(evidence_pack: dict[str, Any], platform_content_id: str) -> dict[str, Any]:
     evidence = copy.deepcopy(evidence_pack)
     evidence["discovered_platform_content_id"] = platform_content_id

tests/test_walk_graph_config.py

@@ -38,6 +38,17 @@ def test_policy_pins_gemini_cap_and_workers():
     assert policy_global["gemini_max_workers"] == 4
 
 
+def test_policy_allows_smoke_env_overrides(monkeypatch):
+    monkeypatch.setenv("CONTENT_AGENT_WALK_MAX_TOTAL_ACTIONS_PER_RUN", "4")
+    monkeypatch.setenv("CONTENT_AGENT_GEMINI_MAX_WORKERS", "3")
+
+    policy_global = WalkGraphStore().load_policy()["global"]
+
+    assert policy_global["max_total_actions_per_run"] == 4
+    assert policy_global["gemini_max_workers"] == 3
+    assert policy_global["gemini_calls_per_run_cap"] == 200
+
+
 def test_policy_edge_budgets_match_decided_values():
     # 基线=v1 实际硬限;R7 放宽拍板(2026-06-12):tag 1→3、作者 2→3(真跑顶格实证)。
     budgets = WalkGraphStore().load_policy()["edge_budgets_by_id"]