from __future__ import annotations

import json
import time
from pathlib import Path
from typing import Any

import config_manager

from src.auto_governance import load_auto_governance_snapshot
from src.metrics_agg import aggregate_exec_stats, aggregate_ledger_stats
from src.phase_readiness import build_phase_readiness
from src.rollout_manager import PROMOTION_ORDER, evaluate_promotion_gates, get_promotion_mode, set_promotion_mode
from src.runtime_state.training_review_state import build_training_review_state, public_training_review_state
from src.tenant_manager import load_runtime_state, tenant_paths
from src.training_candidate_value import evaluate_training_candidate_value
from src.training_quality_snapshot import read_training_quality_snapshot


STATE_COLLECT_EVIDENCE = "COLLECT_EVIDENCE"
STATE_PENDING_APPROVAL = "PENDING_APPROVAL"
STATE_MANUAL_REVIEW = "MANUAL_REVIEW"
STATE_PROMOTION_BLOCKED = "PROMOTION_BLOCKED"
STATE_READY_TO_PROMOTE = "READY_TO_PROMOTE"
STATE_ROLLBACK_REQUIRED = "ROLLBACK_REQUIRED"
STATE_FROZEN = "FROZEN"


def _read_json(path: Path) -> dict[str, Any]:
    if not path.exists():
        return {}
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except Exception:
        return {}


def _parse_ts(raw: Any) -> float | None:
    if raw in (None, ""):
        return None
    if isinstance(raw, (int, float)):
        return float(raw)
    try:
        text = str(raw).replace("Z", "+00:00")
        return float(__import__("datetime").datetime.fromisoformat(text).timestamp())
    except Exception:
        return None


def _age_seconds(raw: Any) -> float | None:
    ts = _parse_ts(raw)
    if ts is None:
        return None
    return max(0.0, time.time() - ts)


def _coordinator_paths(project_root: Path, tenant_id: str) -> dict[str, Path]:
    paths = tenant_paths(project_root, tenant_id)
    ops_dir = paths["runtime"].parent
    logs_dir = paths["audit"].parent
    return {
        "state": ops_dir / "evolution_coordinator_state.json",
        "latest": logs_dir / "evolution_coordinator_latest.json",
        "history": logs_dir / "evolution_coordinator_history.ndjson",
    }


def _write_json(path: Path, payload: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")


def _append_history(path: Path, payload: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with open(path, "a", encoding="utf-8") as f:
        f.write(json.dumps(payload, ensure_ascii=False) + "\n")


def _resolve_file_layout() -> dict[str, Path]:
    app_dir = config_manager.get_app_dir()
    config_dir = config_manager.get_config_dir()
    logs_dir = config_manager.get_logs_dir()
    data_dir = config_manager.get_data_dir()
    return {
        "app_dir": app_dir,
        "config_dir": config_dir,
        "logs_dir": logs_dir,
        "data_dir": data_dir,
        "suggestion_file": config_dir / "params_suggestion.json",
        "shadow_file": config_dir / "params_shadow.json",
        "effective_file": config_manager.effective_config_path(),
        "report_file": logs_dir / "evolution_report_latest.json",
        "learner_report_file": data_dir / "offline" / "param_learner_report_latest.json",
        "quality_snapshot_file": data_dir / "offline" / "training_quality_snapshot.json",
        "shadow_summary_file": logs_dir / "shadow_learner_summary_latest.json",
        "live_summary_file": logs_dir / "live_learner_summary_latest.json",
    }


def _fallback_exec_stats(app_dir: Path, logs_dir: Path, tenant_id: str) -> dict[str, Any]:
    exec_stats = aggregate_exec_stats(logs_dir)
    try:
        from src.metrics_agg import aggregate_goal_targets
        goal_targets = aggregate_goal_targets(app_dir, tenant_id)
        month_window = goal_targets.get("window_month", {}) if isinstance(goal_targets, dict) else {}
        closed_trade_count = int(month_window.get("closed_trade_count", 0) or 0)
        profit_factor = float(month_window.get("profit_factor", 0.0) or 0.0)
        if float(exec_stats.get("min_paper_trades", 0) or 0) <= 0 and closed_trade_count > 0:
            exec_stats["min_paper_trades"] = closed_trade_count
        if float(exec_stats.get("min_profit_factor", 0.0) or 0.0) <= 0 and profit_factor > 0:
            exec_stats["min_profit_factor"] = round(profit_factor, 4)
    except Exception:
        pass
    return exec_stats


def _normalize_phase_readiness(app_dir: Path) -> dict[str, Any]:
    try:
        return build_phase_readiness(app_dir)  # type: ignore[arg-type]
    except TypeError:
        try:
            return build_phase_readiness()
        except Exception:
            return {}
    except Exception:
        return {}


def _safe_float(value: Any, default: float = 0.0) -> float:
    try:
        return float(value)
    except Exception:
        return default


def _safe_int(value: Any, default: int = 0) -> int:
    try:
        return int(value)
    except Exception:
        return default


def _clamp(value: float, lower: float, upper: float) -> float:
    return max(lower, min(upper, value))


def _normalize_rate_pct(value: Any) -> float:
    raw = _safe_float(value, 0.0)
    if 0.0 <= raw <= 1.0:
        raw *= 100.0
    return round(raw, 2)


def _extract_first_number(payload: dict[str, Any], keys: list[str], default: float = 0.0) -> float:
    for key in keys:
        if key in payload:
            return _safe_float(payload.get(key), default)
    return default


def _extract_first_int(payload: dict[str, Any], keys: list[str], default: int = 0) -> int:
    for key in keys:
        if key in payload:
            return _safe_int(payload.get(key), default)
    return default


def _normalize_bucket_group(raw_group: Any) -> dict[str, dict[str, Any]]:
    if not isinstance(raw_group, dict):
        return {}
    normalized: dict[str, dict[str, Any]] = {}
    for name, stats in raw_group.items():
        if not isinstance(stats, dict):
            continue
        normalized[str(name)] = {
            "trades": _extract_first_int(stats, ["trades", "n", "count", "total_trades"], 0),
            "win_rate_pct": _normalize_rate_pct(
                stats.get("win_rate_pct", stats.get("win_rate", stats.get("wr", 0.0)))
            ),
            "total_pnl": _safe_float(stats.get("total_pnl", stats.get("pnl", 0.0)), 0.0),
            "profit_factor": _safe_float(stats.get("profit_factor", stats.get("rr_ratio", 0.0)), 0.0),
        }
    return normalized


def _extract_performance_snapshot(payload: dict[str, Any], *, label: str) -> dict[str, Any]:
    raw = payload if isinstance(payload, dict) else {}
    if isinstance(raw.get("offline_stats"), dict):
        merged = {**raw, **raw.get("offline_stats", {})}
    elif isinstance(raw.get("summary"), dict):
        merged = {**raw, **raw.get("summary", {})}
    elif isinstance(raw.get("candidate"), dict):
        merged = {**raw, **raw.get("candidate", {})}
    elif isinstance(raw.get("baseline"), dict):
        merged = {**raw, **raw.get("baseline", {})}
    else:
        merged = dict(raw)

    by_regime = merged.get("by_regime", merged.get("regime", {}))
    by_side = merged.get("by_side", merged.get("side", {}))

    snapshot = {
        "label": label,
        "sample_count": _extract_first_int(
            merged,
            ["sample_count", "real_trades", "total_trades", "trade_count", "trades", "n", "usable_trade_count"],
            0,
        ),
        "win_rate_pct": _normalize_rate_pct(
            merged.get("win_rate_pct", merged.get("win_rate", merged.get("baseline_win_rate", 0.0)))
        ),
        "total_pnl": _safe_float(merged.get("total_pnl", merged.get("pnl", merged.get("avg_pnl", 0.0))), 0.0),
        "profit_factor": _safe_float(merged.get("profit_factor", merged.get("rr_ratio", merged.get("pf", 0.0))), 0.0),
        "max_drawdown_pct": _safe_float(merged.get("max_drawdown_pct", merged.get("drawdown_pct", 0.0)), 0.0),
        "learner_live_rate": _safe_float(merged.get("learnerLiveRate", merged.get("learner_live_rate", 0.0)), 0.0),
        "divergence_count": _safe_int(merged.get("divergenceCount", merged.get("divergence_count", 0)), 0),
        "by_regime": _normalize_bucket_group(by_regime),
        "by_side": _normalize_bucket_group(by_side),
    }
    return snapshot


def _bucket_consistency_penalty(snapshot: dict[str, Any]) -> tuple[float, list[str]]:
    penalties = 0.0
    reasons: list[str] = []
    for group_name in ("by_regime", "by_side"):
        group = snapshot.get(group_name, {})
        if not isinstance(group, dict):
            continue
        for bucket_name, stats in group.items():
            if not isinstance(stats, dict):
                continue
            trades = _extract_first_int(stats, ["trades"], 0)
            if trades < 5:
                continue
            win_rate_pct = _normalize_rate_pct(stats.get("win_rate_pct", 0.0))
            total_pnl = _safe_float(stats.get("total_pnl", 0.0), 0.0)
            if win_rate_pct < 42.0 or total_pnl < 0.0:
                penalties += 0.06
                reasons.append(f"{group_name}:{bucket_name}_weak")
    return min(penalties, 0.18), reasons


def evaluate_promotion_verdict(
    *,
    effective_params: dict[str, Any],
    evolution_report: dict[str, Any],
    shadow_learner: dict[str, Any],
    live_learner: dict[str, Any],
    runtime_state: dict[str, Any],
    exec_stats: dict[str, Any],
    ledger_stats: dict[str, Any],
    dynamic_trigger: dict[str, Any],
    review_pending: bool,
) -> dict[str, Any]:
    baseline_raw = (
        live_learner.get("baseline", {})
        if isinstance(live_learner.get("baseline", {}), dict)
        else shadow_learner.get("baseline", {})
        if isinstance(shadow_learner.get("baseline", {}), dict)
        else {}
    )
    if not baseline_raw:
        baseline_raw = {
            "win_rate": evolution_report.get("baseline_win_rate", 0.0),
            "total_trades": ((evolution_report.get("offline_stats", {}) or {}).get("real_trades", 0)),
            "profit_factor": ((effective_params.get("current", {}) or {}).get("stats", {}) or {}).get("profit_factor", 0.0),
            "total_pnl": ((effective_params.get("current", {}) or {}).get("stats", {}) or {}).get("total_pnl", 0.0),
            "max_drawdown_pct": ((effective_params.get("current", {}) or {}).get("stats", {}) or {}).get("max_drawdown_pct", 0.0),
        }

    candidate_source = "offline"
    candidate_raw: dict[str, Any] = {}
    if isinstance(live_learner, dict) and live_learner:
        candidate_raw = live_learner
        candidate_source = "live"
    elif isinstance(shadow_learner, dict) and shadow_learner:
        candidate_raw = shadow_learner
        candidate_source = "shadow"
    elif isinstance(evolution_report.get("offline_stats", {}), dict):
        candidate_raw = evolution_report.get("offline_stats", {})
        candidate_source = "offline"
    else:
        candidate_raw = ((effective_params.get("current", {}) or {}).get("stats", {}) or {})
        candidate_source = "effective_current"

    baseline = _extract_performance_snapshot(baseline_raw, label="baseline")
    candidate = _extract_performance_snapshot(candidate_raw, label=candidate_source)

    win_rate_delta = round(candidate["win_rate_pct"] - baseline["win_rate_pct"], 2)
    total_pnl_delta = round(candidate["total_pnl"] - baseline["total_pnl"], 2)
    baseline_pnl_abs = max(abs(baseline["total_pnl"]), 1.0)
    pnl_delta_ratio = round(total_pnl_delta / baseline_pnl_abs, 4)
    profit_factor_delta = round(candidate["profit_factor"] - baseline["profit_factor"], 3)
    drawdown_delta = round(candidate["max_drawdown_pct"] - baseline["max_drawdown_pct"], 2)

    bootstrap_mode = candidate["sample_count"] == 0 and baseline["sample_count"] == 0
    score = 0.50
    reasons: list[str] = []
    penalties: list[str] = []

    if bootstrap_mode:
        score += 0.05
        reasons.append("bootstrap_mode_zero_samples")
    elif candidate["sample_count"] >= 40:
        score += 0.10
        reasons.append(f"sample_count strong ({candidate['sample_count']})")
    elif candidate["sample_count"] >= 20:
        score += 0.05
        reasons.append(f"sample_count adequate ({candidate['sample_count']})")
    else:
        score -= 0.15
        penalties.append(f"sample_count weak ({candidate['sample_count']})")

    if win_rate_delta >= 5.0:
        score += 0.18
        reasons.append(f"win_rate improved by {win_rate_delta:.1f}pt")
    elif win_rate_delta >= 2.0:
        score += 0.10
        reasons.append(f"win_rate modestly improved by {win_rate_delta:.1f}pt")
    elif win_rate_delta <= -5.0:
        score -= 0.22
        penalties.append(f"win_rate deteriorated by {win_rate_delta:.1f}pt")
    elif win_rate_delta <= -2.0:
        score -= 0.12
        penalties.append(f"win_rate slightly deteriorated by {win_rate_delta:.1f}pt")

    if pnl_delta_ratio >= 0.20:
        score += 0.12
        reasons.append(f"total_pnl improved by {pnl_delta_ratio*100:.0f}%")
    elif pnl_delta_ratio <= -0.20:
        score -= 0.16
        penalties.append(f"total_pnl deteriorated by {pnl_delta_ratio*100:.0f}%")

    if profit_factor_delta >= 0.20:
        score += 0.12
        reasons.append(f"profit_factor improved by {profit_factor_delta:.2f}")
    elif profit_factor_delta <= -0.15:
        score -= 0.16
        penalties.append(f"profit_factor deteriorated by {profit_factor_delta:.2f}")

    if baseline["max_drawdown_pct"] > 0 or candidate["max_drawdown_pct"] > 0:
        if drawdown_delta <= -2.0:
            score += 0.12
            reasons.append(f"drawdown improved by {abs(drawdown_delta):.1f}pt")
        elif drawdown_delta >= 2.0:
            score -= 0.18
            penalties.append(f"drawdown worsened by {drawdown_delta:.1f}pt")

    consistency_penalty, consistency_reasons = _bucket_consistency_penalty(candidate)
    if consistency_penalty > 0:
        score -= consistency_penalty
        penalties.extend(consistency_reasons)
    elif candidate["by_regime"] or candidate["by_side"]:
        score += 0.04
        reasons.append("regime_side_consistency_ok")

    divergence_count = candidate.get("divergence_count", 0)
    learner_live_rate = candidate.get("learner_live_rate", 0.0)
    if divergence_count >= 3:
        score -= 0.12
        penalties.append(f"shadow_divergence_high:{divergence_count}")
    if candidate_source == "live" and learner_live_rate < 50.0:
        score -= 0.08
        penalties.append(f"learner_live_rate_low:{learner_live_rate:.1f}")

    intraday_drawdown_pct = abs(_safe_float(runtime_state.get("intraday_drawdown_pct", 0.0)))
    daily_loss_pct = abs(_safe_float(runtime_state.get("daily_loss_pct", 0.0)))
    consecutive_losses = _safe_int(runtime_state.get("consecutive_losses", 0))
    if intraday_drawdown_pct >= 2.0 or daily_loss_pct >= 1.5 or consecutive_losses >= 3:
        score -= 0.12
        penalties.append("runtime_risk_elevated")

    exec_reject_rate_pct = _safe_float(exec_stats.get("exec_reject_rate_pct", 0.0))
    exec_timeout_rate_pct = _safe_float(exec_stats.get("exec_timeout_rate_pct", 0.0))
    decision_error_rate_pct_1h = _safe_float(ledger_stats.get("decision_error_rate_pct_1h", 0.0))
    if exec_reject_rate_pct >= 20.0 or exec_timeout_rate_pct >= 10.0 or decision_error_rate_pct_1h >= 15.0:
        score -= 0.10
        penalties.append("execution_or_decision_instability")

    if dynamic_trigger.get("collectMoreEvidence", False):
        score -= 0.06
        penalties.append("dynamic_trigger_requires_more_evidence")
    if dynamic_trigger.get("repairRequired", False):
        score -= 0.10
        penalties.append("dynamic_trigger_requires_repair")
    if dynamic_trigger.get("freezeReasons"):
        score -= 0.20
        penalties.append("dynamic_trigger_frozen")

    score = round(_clamp(score, 0.0, 1.0), 3)
    if bootstrap_mode:
        confidence = "medium"
    else:
        confidence = "high" if candidate["sample_count"] >= 40 else "medium" if candidate["sample_count"] >= 20 else "low"

    hard_negative = (
        False
        if bootstrap_mode
        else (
            (candidate["sample_count"] >= 10 and win_rate_delta <= -5.0)
            or profit_factor_delta <= -0.20
            or drawdown_delta >= 3.0
            or (candidate["total_pnl"] < 0 and baseline["total_pnl"] >= 0 and candidate["sample_count"] >= 10)
            or divergence_count >= 4
        )
    )
    if bootstrap_mode:
        promote_allowed = (
            score >= 0.40
            and not dynamic_trigger.get("repairRequired", False)
            and not dynamic_trigger.get("freezeReasons")
        )
    else:
        promote_allowed = (
            score >= 0.67
            and confidence != "low"
            and candidate["sample_count"] >= 20
            and win_rate_delta >= 2.0
            and profit_factor_delta >= 0.0
            and drawdown_delta <= 1.0
            and not dynamic_trigger.get("repairRequired", False)
            and not dynamic_trigger.get("freezeReasons")
        )
    rollback_required = bool(review_pending and hard_negative)

    if rollback_required:
        verdict = "ROLLBACK"
    elif promote_allowed:
        verdict = "PROMOTE"
    else:
        verdict = "HOLD"

    summary_reasons = reasons + penalties
    if not summary_reasons:
        summary_reasons.append("no_strong_evidence")

    return {
        "verdict": verdict,
        "promotionScore": score,
        "confidence": confidence,
        "promoteAllowed": promote_allowed,
        "rollbackRequired": rollback_required,
        "candidateSource": candidate_source,
        "baseline": baseline,
        "candidate": candidate,
        "deltas": {
            "winRateDeltaPt": win_rate_delta,
            "totalPnlDelta": total_pnl_delta,
            "totalPnlDeltaRatio": pnl_delta_ratio,
            "profitFactorDelta": profit_factor_delta,
            "maxDrawdownDeltaPt": drawdown_delta,
        },
        "reasons": summary_reasons[:12],
    }


def evaluate_dynamic_trigger_thresholds(
    *,
    quality_context: dict[str, Any],
    candidate_value: dict[str, Any],
    auto_governance: dict[str, Any],
    runtime_state: dict[str, Any],
    exec_stats: dict[str, Any],
    ledger_stats: dict[str, Any],
    goal_gates: dict[str, Any],
    has_training_output: bool,
    suggestion_exists: bool,
    effective_exists: bool,
) -> dict[str, Any]:
    usable_trade_count = _safe_int(quality_context.get("usable_trade_count", 0))
    closure_coverage_pct = _safe_float(quality_context.get("closure_coverage_pct", 0.0))
    natural_closed_ratio_pct = _safe_float(quality_context.get("natural_closed_ratio_pct", 0.0))
    critical_anomalies = _safe_int(quality_context.get("critical_anomalies", 0))
    candidate_score = _safe_float(candidate_value.get("candidate_score", 0.0))
    ready_for_training = bool(quality_context.get("ready_for_training", False))
    blocked_reasons = quality_context.get("blocked_reasons", []) if isinstance(quality_context.get("blocked_reasons", []), list) else []

    intraday_drawdown_pct = abs(_safe_float(runtime_state.get("intraday_drawdown_pct", 0.0)))
    daily_loss_pct = abs(_safe_float(runtime_state.get("daily_loss_pct", 0.0)))
    consecutive_losses = _safe_int(runtime_state.get("consecutive_losses", 0))
    kill_switch = bool(runtime_state.get("kill_switch", False))

    exec_reject_rate_pct = _safe_float(exec_stats.get("exec_reject_rate_pct", 0.0))
    exec_timeout_rate_pct = _safe_float(exec_stats.get("exec_timeout_rate_pct", 0.0))
    decision_error_rate_pct_1h = _safe_float(ledger_stats.get("decision_error_rate_pct_1h", 0.0))
    governance_status = str(auto_governance.get("status", "NOOP") or "NOOP").upper()
    governance_next_action = str(auto_governance.get("next_action", "NONE") or "NONE").upper()
    goal_all_passed = bool(goal_gates.get("all_passed", True))

    bootstrap_mode = usable_trade_count == 0 and not (
        intraday_drawdown_pct >= 2.0 or daily_loss_pct >= 1.5 or consecutive_losses >= 3
    )
    min_trades = 0 if bootstrap_mode else 10
    min_closure_pct = 0.0 if bootstrap_mode else 20.0
    min_natural_ratio_pct = 0.0 if bootstrap_mode else 40.0
    min_candidate_score = 0.15 if bootstrap_mode else 0.30
    max_critical_anomalies = 1
    adjustments: list[str] = []
    blockers: list[str] = []
    supportive_signals: list[str] = []
    freeze_reasons: list[str] = []
    repair_reasons: list[str] = []
    if bootstrap_mode:
        adjustments.append("bootstrap_mode_zero_trades")

    if not bootstrap_mode:
        if intraday_drawdown_pct >= 2.0 or daily_loss_pct >= 1.5 or consecutive_losses >= 3:
            min_trades += 10
            min_closure_pct += 10.0
            min_natural_ratio_pct += 10.0
            min_candidate_score += 0.10
            max_critical_anomalies = 0
            adjustments.append("runtime_risk_elevated")
        if exec_reject_rate_pct >= 20.0 or exec_timeout_rate_pct >= 10.0 or decision_error_rate_pct_1h >= 15.0:
            min_trades += 8
            min_closure_pct += 5.0
            min_candidate_score += 0.05
            adjustments.append("execution_instability_detected")
        if governance_status in {"BLOCKED", "FREEZE", "ROLLBACK", "STOP"} or governance_next_action in {"FREEZE", "ROLLBACK"}:
            min_trades += 5
            min_closure_pct += 5.0
            min_candidate_score += 0.05
            adjustments.append("governance_caution_mode")
        if usable_trade_count >= 60 and closure_coverage_pct >= 60.0 and critical_anomalies == 0:
            min_candidate_score = max(0.20, min_candidate_score - 0.05)
            adjustments.append("high_quality_sample_credit")

    if bootstrap_mode:
        min_trades = 0
        min_closure_pct = 0.0
        min_natural_ratio_pct = 0.0
        min_candidate_score = round(_clamp(min_candidate_score, 0.10, 0.75), 2)
    else:
        min_trades = int(_clamp(float(min_trades), 10.0, 40.0))
        min_closure_pct = round(_clamp(min_closure_pct, 20.0, 60.0), 1)
        min_natural_ratio_pct = round(_clamp(min_natural_ratio_pct, 40.0, 80.0), 1)
        min_candidate_score = round(_clamp(min_candidate_score, 0.25, 0.75), 2)

    if kill_switch:
        freeze_reasons.append(str(runtime_state.get("kill_switch_reason", "kill_switch_active") or "kill_switch_active"))
    if not goal_all_passed:
        freeze_reasons.append("goal_gates_failed")

    if not ready_for_training and not bootstrap_mode:
        blockers.extend([f"quality_gate:{reason}" for reason in blocked_reasons] or ["quality_gate_blocked"])
    if bootstrap_mode:
        supportive_signals.append("bootstrap_skip_quality_gate")
    if usable_trade_count < min_trades:
        blockers.append(f"trade_count:{usable_trade_count}<{min_trades}")
    else:
        supportive_signals.append(f"trade_count_ok:{usable_trade_count}>={min_trades}")
    if closure_coverage_pct < min_closure_pct:
        blockers.append(f"closure_coverage:{closure_coverage_pct}<{min_closure_pct}")
    else:
        supportive_signals.append(f"closure_coverage_ok:{closure_coverage_pct}>={min_closure_pct}")
    if natural_closed_ratio_pct < min_natural_ratio_pct:
        blockers.append(f"natural_ratio:{natural_closed_ratio_pct}<{min_natural_ratio_pct}")
    else:
        supportive_signals.append(f"natural_ratio_ok:{natural_closed_ratio_pct}>={min_natural_ratio_pct}")
    if critical_anomalies > max_critical_anomalies:
        blockers.append(f"critical_anomalies:{critical_anomalies}>{max_critical_anomalies}")
        repair_reasons.append("critical_anomalies_above_dynamic_limit")
    else:
        supportive_signals.append(f"critical_anomalies_ok:{critical_anomalies}<={max_critical_anomalies}")
    if candidate_score < min_candidate_score:
        blockers.append(f"candidate_score:{candidate_score}<{min_candidate_score}")
    else:
        supportive_signals.append(f"candidate_score_ok:{candidate_score}>={min_candidate_score}")

    if exec_reject_rate_pct >= 20.0:
        repair_reasons.append("exec_reject_rate_high")
    if exec_timeout_rate_pct >= 10.0:
        repair_reasons.append("exec_timeout_rate_high")
    if decision_error_rate_pct_1h >= 15.0:
        repair_reasons.append("decision_error_rate_high")
    if blocked_reasons and not bootstrap_mode:
        repair_reasons.extend([f"quality_gate:{reason}" for reason in blocked_reasons])

    sample_score = _clamp(usable_trade_count / max(float(min_trades), 1.0), 0.0, 1.0)
    closure_score = _clamp(closure_coverage_pct / max(min_closure_pct, 1.0), 0.0, 1.0)
    natural_score = _clamp(natural_closed_ratio_pct / max(min_natural_ratio_pct, 1.0), 0.0, 1.0)
    candidate_metric_score = _clamp(candidate_score / max(min_candidate_score, 0.01), 0.0, 1.0)
    anomaly_score = 1.0 if critical_anomalies <= max_critical_anomalies else 0.0
    penalty = 0.0
    if intraday_drawdown_pct >= 2.0 or daily_loss_pct >= 1.5:
        penalty += 0.15
    if exec_reject_rate_pct >= 20.0 or exec_timeout_rate_pct >= 10.0:
        penalty += 0.10
    if decision_error_rate_pct_1h >= 15.0:
        penalty += 0.08
    if governance_status in {"BLOCKED", "FREEZE", "ROLLBACK", "STOP"}:
        penalty += 0.10
    if not goal_all_passed:
        penalty += 0.20

    trigger_score = _clamp(
        (sample_score * 0.30)
        + (closure_score * 0.25)
        + (natural_score * 0.15)
        + (candidate_metric_score * 0.20)
        + (anomaly_score * 0.10)
        - penalty,
        0.0,
        1.0,
    )
    confidence_band = "high" if trigger_score >= 0.75 else "medium" if trigger_score >= 0.50 else "low"
    evidence_ready = (
        usable_trade_count >= min_trades
        and closure_coverage_pct >= min_closure_pct
        and natural_closed_ratio_pct >= min_natural_ratio_pct
        and critical_anomalies <= max_critical_anomalies
    )
    training_trigger_ready = evidence_ready and ready_for_training and not freeze_reasons
    approval_trigger_ready = training_trigger_ready and candidate_score >= min_candidate_score and trigger_score >= 0.55
    suggestion_trigger_ready = training_trigger_ready and has_training_output and not suggestion_exists and not effective_exists and trigger_score >= 0.55
    run_training_ready = training_trigger_ready and not has_training_output and trigger_score >= 0.60
    repair_required = bool(repair_reasons) and not freeze_reasons
    collect_more_evidence = not freeze_reasons and not repair_required and not approval_trigger_ready

    if freeze_reasons:
        recommended_next_action = "freeze_and_monitor"
    elif repair_required:
        recommended_next_action = "repair_blockers"
    elif run_training_ready:
        recommended_next_action = "run_offline_training"
    elif suggestion_trigger_ready:
        recommended_next_action = "generate_suggestion"
    elif approval_trigger_ready and suggestion_exists:
        recommended_next_action = "approve_to_shadow"
    elif collect_more_evidence:
        recommended_next_action = "collect_more_evidence"
    else:
        recommended_next_action = "wait_for_manual_review"

    return {
        "dynamicThresholds": {
            "minTrades": min_trades,
            "minClosureCoveragePct": min_closure_pct,
            "minNaturalClosedRatioPct": min_natural_ratio_pct,
            "minCandidateScore": min_candidate_score,
            "maxCriticalAnomalies": max_critical_anomalies,
        },
        "observedMetrics": {
            "usableTradeCount": usable_trade_count,
            "closureCoveragePct": closure_coverage_pct,
            "naturalClosedRatioPct": natural_closed_ratio_pct,
            "criticalAnomalies": critical_anomalies,
            "candidateScore": candidate_score,
            "intradayDrawdownPct": intraday_drawdown_pct,
            "dailyLossPct": daily_loss_pct,
            "consecutiveLosses": consecutive_losses,
            "execRejectRatePct": exec_reject_rate_pct,
            "execTimeoutRatePct": exec_timeout_rate_pct,
            "decisionErrorRatePct1h": decision_error_rate_pct_1h,
        },
        "adjustments": adjustments,
        "blockers": blockers,
        "supportiveSignals": supportive_signals,
        "repairReasons": sorted(set(repair_reasons)),
        "freezeReasons": freeze_reasons,
        "triggerScore": round(trigger_score, 3),
        "confidenceBand": confidence_band,
        "evidenceReady": evidence_ready,
        "trainingTriggerReady": training_trigger_ready,
        "runTrainingReady": run_training_ready,
        "suggestionTriggerReady": suggestion_trigger_ready,
        "approvalTriggerReady": approval_trigger_ready,
        "repairRequired": repair_required,
        "collectMoreEvidence": collect_more_evidence,
        "recommendedNextAction": recommended_next_action,
        "hasTrainingOutput": has_training_output,
    }


def _determine_decision(
    *,
    data: dict[str, Any],
    candidate_value: dict[str, Any],
    promotion_verdict: dict[str, Any],
    review_state: dict[str, Any],
    quality_context: dict[str, Any],
    dynamic_trigger: dict[str, Any],
    promotion_mode: str,
    gate_readiness: list[dict[str, Any]],
    runtime_state: dict[str, Any],
    external_inputs: dict[str, Any],
) -> tuple[str, dict[str, Any]]:
    suggestion = data.get("suggestion", {})
    effective = data.get("effectiveParams", {})
    shadow_exists = bool(data.get("shadowExists", False))
    review_pending = bool(review_state.get("trainingReviewPendingRaw", False))
    candidate_ready = bool(candidate_value.get("ready_for_manual_review", False))
    verdict_promote_allowed = bool(promotion_verdict.get("promoteAllowed", False))
    verdict_rollback_required = bool(promotion_verdict.get("rollbackRequired", False))
    ready_for_training = bool(quality_context.get("ready_for_training", False))
    blocked_reasons = quality_context.get("blocked_reasons", []) if isinstance(quality_context, dict) else []
    bootstrap_mode = bool(dynamic_trigger.get("adjustments", []) and "bootstrap_mode_zero_trades" in dynamic_trigger.get("adjustments", []))
    trigger_blockers = dynamic_trigger.get("blockers", []) if isinstance(dynamic_trigger.get("blockers", []), list) else []
    trigger_repair_required = bool(dynamic_trigger.get("repairRequired", False))
    approval_trigger_ready = bool(dynamic_trigger.get("approvalTriggerReady", False))
    suggestion_trigger_ready = bool(dynamic_trigger.get("suggestionTriggerReady", False))
    run_training_ready = bool(dynamic_trigger.get("runTrainingReady", False))
    unmet_gates = [g for g in gate_readiness if not g.get("ready", False)]
    next_mode = gate_readiness[0].get("next_mode") if gate_readiness else None
    goal_gates = external_inputs.get("goal_gates", {}) if isinstance(external_inputs.get("goal_gates", {}), dict) else {}
    goal_all_passed = bool(goal_gates.get("all_passed", True))
    evolution_verdict = (
        external_inputs.get("evolution_verdict", {})
        if isinstance(external_inputs.get("evolution_verdict", {}), dict)
        else {}
    )
    evolution_status = str(evolution_verdict.get("status", "INSUFFICIENT_EVIDENCE") or "INSUFFICIENT_EVIDENCE")

    missing_conditions: list[str] = []
    recommended_actions: list[str] = []
    forbidden_actions: list[str] = []
    freeze_reasons: list[str] = []
    next_step = "等待新的参数建议"

    if bool(runtime_state.get("kill_switch", False)):
        freeze_reasons.append(str(runtime_state.get("kill_switch_reason", "kill_switch_active") or "kill_switch_active"))
    if not goal_all_passed:
        freeze_reasons.append("goal_gates_failed")
    freeze_reasons.extend([str(reason) for reason in dynamic_trigger.get("freezeReasons", []) if str(reason)])

    if freeze_reasons:
        coordinator_state = STATE_FROZEN
        decision_status = "FROZEN"
        recommended_actions = ["freeze_and_monitor", "inspect_runtime_state"]
        forbidden_actions = ["approve_to_shadow", "complete_manual_review", "promote_to_live"]
        next_step = "目标门槛或运行态风控未通过，冻结演化推进，仅保留监控与基线保护"
    elif review_pending and (not candidate_ready or verdict_rollback_required):
        coordinator_state = STATE_ROLLBACK_REQUIRED
        decision_status = "ROLLBACK_REQUIRED"
        if not candidate_ready:
            missing_conditions.append("candidate_value_insufficient")
            missing_conditions.extend([f"candidate_value:{reason}" for reason in candidate_value.get("reasons", [])])
        if verdict_rollback_required:
            missing_conditions.extend([f"promotion_verdict:{reason}" for reason in promotion_verdict.get("reasons", [])])
        recommended_actions = ["rollback_invalid_candidate"]
        forbidden_actions = ["complete_manual_review", "promote_to_live", "advance_promotion_mode"]
        next_step = "当前观察态候选未通过价值门槛，必须先回滚到稳定参数"
    elif suggestion and not review_pending:
        coordinator_state = STATE_PENDING_APPROVAL
        if approval_trigger_ready and candidate_ready and not verdict_rollback_required:
            decision_status = "WAIT_APPROVAL"
            recommended_actions = ["approve_to_shadow", "reject_invalid_candidate"]
            forbidden_actions = ["promote_to_live", "advance_promotion_mode"]
            next_step = "当前有待批建议，动态阈值已通过，可决定是否推进到可回滚观察态"
        else:
            decision_status = "BLOCKED"
            missing_conditions.extend(trigger_blockers)
            missing_conditions.extend([f"promotion_verdict:{reason}" for reason in promotion_verdict.get("reasons", [])])
            recommended_actions = ["reject_invalid_candidate"]
            if trigger_repair_required:
                recommended_actions.extend([f"repair:{reason}" for reason in dynamic_trigger.get("repairReasons", [])])
            else:
                recommended_actions.append("collect_more_evidence")
            forbidden_actions = ["approve_to_shadow", "promote_to_live", "advance_promotion_mode"]
            next_step = "当前虽有建议，但动态触发阈值未满足，禁止进入观察态"
    elif review_pending:
        coordinator_state = STATE_MANUAL_REVIEW
        decision_status = "WAIT_HUMAN_REVIEW"
        missing_conditions.append("manual_review_pending")
        recommended_actions = ["complete_manual_review", "rollback_invalid_candidate"]
        forbidden_actions = ["promote_to_live", "advance_promotion_mode"]
        next_step = "当前处于观察态人工复核阶段，先完成 review 再决定是否晋级"
    else:
        if not suggestion and not effective:
            missing_conditions.append("no_suggestion")
        if not shadow_exists:
            if bootstrap_mode:
                recommended_actions.append("rebuild_shadow_params")
            else:
                missing_conditions.append("shadow_params_missing")
        if not ready_for_training and not bootstrap_mode:
            missing_conditions.extend([f"quality_gate:{r}" for r in blocked_reasons] or ["quality_gate_blocked"])
        if not candidate_ready:
            missing_conditions.extend([f"candidate_value:{reason}" for reason in candidate_value.get("reasons", [])])
        missing_conditions.extend(trigger_blockers)
        if promotion_verdict.get("verdict") == "HOLD":
            missing_conditions.extend([f"promotion_verdict:{reason}" for reason in promotion_verdict.get("reasons", [])])
        if unmet_gates:
            missing_conditions.extend([f"promotion_gate:{g.get('gate')}" for g in unmet_gates])

        if missing_conditions:
            coordinator_state = STATE_PROMOTION_BLOCKED if effective else STATE_COLLECT_EVIDENCE
            decision_status = "BLOCKED" if effective else "NO_ACTION"
            if not suggestion and not effective:
                if suggestion_trigger_ready:
                    recommended_actions.append("generate_or_review_new_suggestion")
                elif run_training_ready:
                    recommended_actions.append("run_offline_training")
                else:
                    recommended_actions.append("collect_more_evidence")
            if not shadow_exists:
                recommended_actions.append("rebuild_shadow_params")
            if not ready_for_training or trigger_repair_required:
                recommended_actions.extend([f"repair:{r}" for r in blocked_reasons] or ["repair_training_quality"])
                recommended_actions.extend([f"repair:{r}" for r in dynamic_trigger.get("repairReasons", [])])
            if unmet_gates:
                recommended_actions.extend([f"wait_or_improve:{g.get('gate')}" for g in unmet_gates])
            forbidden_actions.extend(["promote_to_live", "advance_promotion_mode"])
            next_step = "当前仍有阻断项或晋级门槛未满足，继续收集证据并修复阻断项"
        elif verdict_promote_allowed:
            coordinator_state = STATE_READY_TO_PROMOTE
            decision_status = "READY_TO_PROMOTE"
            recommended_actions.append(f"promote_to_{next_mode or 'next_mode'}")
            next_step = f"已满足观察态与晋级门槛，可推进到 {next_mode or '下一阶段'}"
        else:
            coordinator_state = STATE_PROMOTION_BLOCKED if effective else STATE_COLLECT_EVIDENCE
            decision_status = "BLOCKED" if effective else "NO_ACTION"
            missing_conditions.extend([f"promotion_verdict:{reason}" for reason in promotion_verdict.get("reasons", [])])
            recommended_actions.append("collect_more_evidence")
            forbidden_actions.extend(["promote_to_live", "advance_promotion_mode"])
            next_step = "当前缺少足够的量化改善证据，先继续观察，不允许推进晋级"

    if decision_status in {"FROZEN", "ROLLBACK_REQUIRED"}:
        autonomy_action = "FORCE_BASELINE_GUARD" if decision_status == "ROLLBACK_REQUIRED" else "FREEZE_AND_MONITOR"
        autonomy_notice = "candidate_invalid_require_rollback" if decision_status == "ROLLBACK_REQUIRED" else "goal_gates_failed_keep_shadow"
    elif decision_status == "READY_TO_PROMOTE" and evolution_status == "SIGNIFICANT_OUTPERFORM":
        autonomy_action = "PROMOTION_READY"
        autonomy_notice = "promotion_ready_waiting_governance"
    elif evolution_status == "SIGNIFICANT_UNDERPERFORM":
        autonomy_action = "FORCE_BASELINE_GUARD"
        autonomy_notice = "learner_underperform_hold_baseline"
    else:
        autonomy_action = "CONTINUE_COLLECT_EVIDENCE"
        autonomy_notice = "not_significant_keep_collecting"

    decision = {
        "status": decision_status,
        "currentState": coordinator_state,
        "canPromote": decision_status == "READY_TO_PROMOTE",
        "canRollback": bool((data.get("rollbackState", {}) or {}).get("available", False) and effective),
        "reviewPending": review_pending,
        "trainingReviewPending": review_pending,
        "shadowExists": shadow_exists,
        "promotionMode": promotion_mode,
        "promotionTarget": next_mode,
        "gateReadiness": gate_readiness,
        "missingConditions": missing_conditions,
        "recommendedActions": recommended_actions,
        "forbiddenActions": forbidden_actions,
        "nextStep": next_step,
        "qualityContext": quality_context,
        "candidateValue": candidate_value,
        "promotionVerdict": promotion_verdict,
        "dynamicTrigger": dynamic_trigger,
        "freezeReasons": freeze_reasons,
        "autonomyAction": autonomy_action,
        "autonomyNotice": autonomy_notice,
    }
    return coordinator_state, decision


def build_evolution_coordinator_payload(
    project_root: Path | None = None,
    tenant_id: str = "default",
    source: str = "manual",
    external_inputs: dict[str, Any] | None = None,
) -> dict[str, Any]:
    root = project_root or config_manager.get_app_dir()
    external = external_inputs if isinstance(external_inputs, dict) else {}
    layout = _resolve_file_layout()

    suggestion = _read_json(layout["suggestion_file"])
    applied_suggestion = {}
    effective = config_manager.load_effective_runtime_config()
    evolution_report = _read_json(layout["report_file"])
    param_learner = _read_json(layout["learner_report_file"])
    shadow_learner = _read_json(layout["shadow_summary_file"])
    live_learner = _read_json(layout["live_summary_file"])
    quality_context = read_training_quality_snapshot()
    auto_governance = load_auto_governance_snapshot(layout["app_dir"])
    phase_readiness = _normalize_phase_readiness(layout["app_dir"])
    rollback_state = {}
    try:
        from src.param_flow import get_effective_rollback_state
        rollback_state = get_effective_rollback_state()
    except Exception:
        rollback_state = {}

    candidate_value = evaluate_training_candidate_value(
        quality_snapshot=quality_context,
        bootstrap_mode=(_safe_int(quality_context.get("usable_trade_count", 0)) == 0),
    )
    review_state = build_training_review_state(effective, candidate_value)
    promotion_mode = get_promotion_mode(root, tenant_id)
    exec_stats = _fallback_exec_stats(layout["app_dir"], layout["logs_dir"], tenant_id)
    ledger_stats = aggregate_ledger_stats(layout["logs_dir"])
    gate_readiness = evaluate_promotion_gates(root, tenant_id, exec_stats)
    runtime_state = load_runtime_state(root, tenant_id)
    goal_gates = external.get("goal_gates", {}) if isinstance(external.get("goal_gates", {}), dict) else {}
    has_training_output = bool((layout["data_dir"] / "offline" / "best_params.json").exists())
    dynamic_trigger = evaluate_dynamic_trigger_thresholds(
        quality_context=quality_context,
        candidate_value=candidate_value,
        auto_governance=auto_governance,
        runtime_state=runtime_state,
        exec_stats=exec_stats,
        ledger_stats=ledger_stats,
        goal_gates=goal_gates,
        has_training_output=has_training_output,
        suggestion_exists=bool(suggestion),
        effective_exists=bool(effective),
    )
    promotion_verdict = evaluate_promotion_verdict(
        effective_params=effective,
        evolution_report=evolution_report,
        shadow_learner=shadow_learner,
        live_learner=live_learner,
        runtime_state=runtime_state,
        exec_stats=exec_stats,
        ledger_stats=ledger_stats,
        dynamic_trigger=dynamic_trigger,
        review_pending=bool(review_state.get("trainingReviewPendingRaw", False)),
    )

    data: dict[str, Any] = {
        "success": True,
        "stage": "NO_SUGGESTION",
        "suggestion": suggestion,
        "appliedSuggestion": applied_suggestion,
        "effectiveParams": effective,
        "effectiveOfflineParams": review_state.get("effectiveParamMap", {}),
        "effectiveParamMap": review_state.get("effectiveParamMap", {}),
        "evolutionReport": {
            "timestamp": evolution_report.get("timestamp"),
            "status": evolution_report.get("status"),
            "quality_context": evolution_report.get("quality_context", evolution_report.get("quality_gate", {})),
            "param_learner": evolution_report.get("param_learner", {}),
        } if evolution_report else {},
        "paramLearner": param_learner,
        "shadowLearner": shadow_learner,
        "liveLearner": live_learner,
        "autoGovernance": auto_governance,
        "phaseReadiness": phase_readiness,
        "rollbackState": rollback_state,
        "shadowExists": layout["shadow_file"].exists(),
        "runtimeState": runtime_state,
        "hasTrainingOutput": has_training_output,
        "dynamicTrigger": dynamic_trigger,
        "promotionVerdict": promotion_verdict,
    }

    if suggestion:
        data["stage"] = "PENDING_APPROVAL"
    if review_state.get("trainingReviewPendingRaw", False):
        data["stage"] = "REVIEW_PENDING"

    coordinator_state, decision = _determine_decision(
        data=data,
        candidate_value=candidate_value,
        promotion_verdict=promotion_verdict,
        review_state=review_state,
        quality_context=quality_context,
        dynamic_trigger=dynamic_trigger,
        promotion_mode=promotion_mode,
        gate_readiness=gate_readiness,
        runtime_state=runtime_state,
        external_inputs=external,
    )
    if decision.get("canPromote", False):
        data["stage"] = "READY_TO_PROMOTE"

    approval_summary = {
        **public_training_review_state(review_state),
        "hasEffectiveParams": bool(review_state.get("hasEffectiveParams", False) or effective),
    }

    paths = _coordinator_paths(root, tenant_id)
    prev_state = _read_json(paths["state"])
    state_summary = {
        "updated_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
        "tenant_id": tenant_id or "default",
        "source": source,
        "state": coordinator_state,
        "previous_state": prev_state.get("state", ""),
        "transitioned": str(prev_state.get("state", "")) != coordinator_state,
        "stage": data["stage"],
        "promotion_mode": promotion_mode,
        "promotion_target": decision.get("promotionTarget"),
        "decision_status": decision.get("status"),
        "recommended_actions": decision.get("recommendedActions", []),
        "forbidden_actions": decision.get("forbiddenActions", []),
        "freeze_reasons": decision.get("freezeReasons", []),
        "review_pending": decision.get("reviewPending", False),
        "can_promote": decision.get("canPromote", False),
        "can_rollback": decision.get("canRollback", False),
        "dynamic_trigger_score": dynamic_trigger.get("triggerScore", 0.0),
        "dynamic_trigger_action": dynamic_trigger.get("recommendedNextAction", ""),
        "promotion_verdict": promotion_verdict.get("verdict", ""),
        "promotion_score": promotion_verdict.get("promotionScore", 0.0),
    }

    payload = {
        **data,
        "decision": decision,
        "approvalSummary": approval_summary,
        "coordinator": state_summary,
        "_source_paths": {
            "params_suggestion": str(layout["suggestion_file"]),
            "params_shadow": str(layout["shadow_file"]),
            "effective_params": str(layout["effective_file"]),
            "evolution_report": str(layout["report_file"]),
            "param_learner_report": str(layout["learner_report_file"]),
            "training_quality_snapshot": str(layout["quality_snapshot_file"]),
            "shadow_learner_summary": str(layout["shadow_summary_file"]),
            "live_learner_summary": str(layout["live_summary_file"]),
            "evolution_coordinator_state": str(paths["state"]),
            "evolution_coordinator_latest": str(paths["latest"]),
        },
    }

    _write_json(paths["state"], state_summary)
    _write_json(paths["latest"], payload)
    _append_history(paths["history"], state_summary)
    return payload


class EvolutionCoordinator:
    def __init__(self, project_root: Path | None = None, tenant_id: str = "default"):
        self.project_root = project_root or config_manager.get_app_dir()
        self.tenant_id = tenant_id or "default"

    def snapshot(self, *, source: str = "manual", external_inputs: dict[str, Any] | None = None) -> dict[str, Any]:
        return build_evolution_coordinator_payload(
            project_root=self.project_root,
            tenant_id=self.tenant_id,
            source=source,
            external_inputs=external_inputs,
        )

    def approve_to_shadow(self, *, source: str = "coordinator") -> dict[str, Any]:
        from src.param_flow import promote_suggestion_to_effective

        files = promote_suggestion_to_effective(source=source)
        snapshot = self.snapshot(source=f"{source}:approve_to_shadow")
        return {"success": True, "files": files, "snapshot": snapshot}

    def complete_manual_review(self, *, source: str = "coordinator") -> dict[str, Any]:
        from src.param_flow import complete_effective_review

        files = complete_effective_review(source=source)
        snapshot = self.snapshot(source=f"{source}:complete_manual_review")
        return {"success": True, "files": files, "snapshot": snapshot}

    def reject_suggestion(self, *, source: str = "coordinator") -> dict[str, Any]:
        from src.param_flow import reject_current_suggestion

        files = reject_current_suggestion(source=source)
        snapshot = self.snapshot(source=f"{source}:reject_suggestion")
        return {"success": True, "files": files, "snapshot": snapshot}

    def rollback_candidate(self, *, source: str = "coordinator", reason: str = "") -> dict[str, Any]:
        from src.param_flow import rollback_effective_params

        files = rollback_effective_params(source=source, reason=reason or "coordinator_requested_rollback")
        snapshot = self.snapshot(source=f"{source}:rollback_candidate")
        return {"success": True, "files": files, "snapshot": snapshot}

    def promote(self, *, source: str = "coordinator") -> dict[str, Any]:
        from src.config_snapshot import create_snapshot

        snapshot = self.snapshot(source=f"{source}:pre_promote")
        decision = snapshot.get("decision", {}) if isinstance(snapshot.get("decision", {}), dict) else {}
        if not decision.get("canPromote", False):
            return {"success": False, "message": "当前未满足推进条件", "snapshot": snapshot}

        current_mode = get_promotion_mode(self.project_root, self.tenant_id)
        if current_mode not in PROMOTION_ORDER:
            return {"success": False, "message": f"未知推进模式: {current_mode}", "snapshot": snapshot}
        next_index = PROMOTION_ORDER.index(current_mode) + 1
        if next_index >= len(PROMOTION_ORDER):
            return {"success": False, "message": "已经处于最终推进阶段", "snapshot": snapshot}

        created_snapshot = create_snapshot(
            self.project_root,
            self.tenant_id,
            reason=f"evolution_promote:{current_mode}->{PROMOTION_ORDER[next_index]}",
        )
        runtime_state = load_runtime_state(self.project_root, self.tenant_id)
        runtime_state["last_known_good_tag"] = created_snapshot.get("tag", "")
        runtime_state["last_promotion_source"] = source
        tenant_runtime = tenant_paths(self.project_root, self.tenant_id)["runtime"]
        _write_json(tenant_runtime, runtime_state)
        audit = set_promotion_mode(self.project_root, self.tenant_id, PROMOTION_ORDER[next_index])
        latest = self.snapshot(source=f"{source}:promote")
        return {"success": True, "promotion": audit, "snapshot": latest, "created_snapshot": created_snapshot}
