from __future__ import annotations

import json
import pathlib
import subprocess
import sys
import time
from typing import Any

CURRENT_DIR = pathlib.Path(__file__).resolve().parent
ROOT_DIR = CURRENT_DIR.parent
for candidate in (ROOT_DIR, CURRENT_DIR):
    candidate_str = str(candidate)
    if candidate_str not in sys.path:
        sys.path.insert(0, candidate_str)

import config_manager
from src.market_context import normalize_regime_label
from src.agent_role_router import (
    ROLE_RISK_MANAGER,
    ROLE_TREND_ANALYST,
    call_llm,
)


APP_DIR = config_manager.get_app_dir()
LOGS_DIR = config_manager.get_logs_dir()
DATA_DIR = config_manager.get_data_dir()
PROJECT_STATE_DIR = APP_DIR / "data" / "project_state"
REPORTS_DIR = APP_DIR / "reports"
SYSTEM_DESIGN_CONTRACT_FILE = PROJECT_STATE_DIR / "system_design_contract.json"

HERMES_TRADE_SNAPSHOT_FILE = LOGS_DIR / "hermes_factor_snapshot.json"
HERMES_TRADE_BRIEF_FILE = LOGS_DIR / "hermes_trade_brief.json"
HERMES_MACRO_CONTEXT_FILE = LOGS_DIR / "hermes_macro_context.json"
HERMES_TRAINING_BRIEF_FILE = LOGS_DIR / "hermes_training_brief.json"
HERMES_GOVERNANCE_BRIEF_FILE = LOGS_DIR / "hermes_governance_brief.json"
HERMES_RUNTIME_RULES_FILE = LOGS_DIR / "hermes_runtime_rules.json"
HERMES_WEEKLY_REVIEW_FILE = REPORTS_DIR / "hermes_weekly_review.md"
HERMES_WEEKLY_REVIEW_JSON_FILE = REPORTS_DIR / "hermes_weekly_review.json"
GOAL_PROGRESS_WEEKLY_JSON_FILE = REPORTS_DIR / "goal_progress_weekly.json"
GOAL_PROGRESS_WEEKLY_MD_FILE = REPORTS_DIR / "goal_progress_weekly.md"
GOAL_PROGRESS_MONTHLY_JSON_FILE = REPORTS_DIR / "goal_progress_monthly.json"
GOAL_PROGRESS_MONTHLY_MD_FILE = REPORTS_DIR / "goal_progress_monthly.md"
HERMES_SKILL_REGISTRY_FILE = PROJECT_STATE_DIR / "hermes_skill_registry.json"
HERMES_SKILL_NAME = "gold-weekend-review"
HERMES_SKILL_PATH = pathlib.Path.home() / ".openclaw" / "hermes" / "skills" / HERMES_SKILL_NAME / "SKILL.md"
def _ensure_parent(path: pathlib.Path) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)


def _write_json(path: pathlib.Path, payload: dict[str, Any]) -> pathlib.Path:
    _ensure_parent(path)
    path.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    return path


def _write_text(path: pathlib.Path, content: str) -> pathlib.Path:
    _ensure_parent(path)
    path.write_text(content, encoding="utf-8")
    return path


def _read_json(path: pathlib.Path) -> dict[str, Any]:
    if not path.exists():
        return {}
    # Retry on transient PermissionError (Windows os.replace file locks from concurrent writers)
    for attempt in range(5):
        try:
            payload = json.loads(path.read_text(encoding="utf-8"))
        except json.JSONDecodeError:
            return {}
        except (PermissionError, OSError):
            if attempt < 4:
                time.sleep(0.05 * (2 ** attempt))
                continue
            return {}
        except Exception:
            return {}
        return payload if isinstance(payload, dict) else {}
    return {}


def _read_recent_json(pattern: str, limit: int = 5) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []
    for path in sorted(LOGS_DIR.glob(pattern), key=lambda item: item.stat().st_mtime, reverse=True)[:limit]:
        payload = _read_json(path)
        if payload:
            payload["_file"] = path.name
            rows.append(payload)
    return rows


def _review_item_text(item: Any) -> str:
    if isinstance(item, str):
        return item
    if isinstance(item, dict):
        return " | ".join(
            str(item.get(key, "") or "").strip()
            for key in ("name", "item", "pattern", "message", "detail", "task", "summary", "type")
            if str(item.get(key, "") or "").strip()
        )
    return str(item or "").strip()


def _extract_review_directives(weekly_review: dict[str, Any]) -> dict[str, Any]:
    directives = {
        "avoid_sessions": [],
        "favor_sessions": [],
        "avoid_regimes": [],
        "favor_regimes": [],
        "focus_topics": [],
        "skill_summary": [],
    }
    session_names = {
        "ASIA",
        "ASIA_LONDON_OVERLAP",
        "LONDON",
        "NEW_YORK",
        "LONDON_NY_OVERLAP",
        "ASIA_NOISE",
        "ASIA_DEFENSIVE",
    }
    regime_names = {
        "STRONG_TREND", "WEAK_TREND", "STEALTH_TREND",
        "RANGING", "TRANSITION", "VOLATILITY_SPIKE", "FAKE_TREND", "EXTREME",
        "TRENDING", "UNCERTAIN", "BREAKOUT", "SHOCK_REPRICING", "EXTREME_TREND",
    }
    positive_markers = ("BEST", "FAVOR", "WIN", "PROFIT", "GOOD", "PROMOTE")
    negative_markers = ("AVOID", "FAIL", "LOSS", "BLOCK", "UNDERPERFORM", "RISK")

    texts: list[str] = []
    for key in ("best_patterns", "failure_patterns", "action_items", "memory_notes", "governance_notes"):
        values = weekly_review.get(key, [])
        if isinstance(values, list):
            texts.extend(_review_item_text(item) for item in values)
    skill = weekly_review.get("skill", {}) if isinstance(weekly_review.get("skill", {}), dict) else {}
    skill_summary = skill.get("summary", []) if isinstance(skill.get("summary", []), list) else []
    directives["skill_summary"] = [str(item) for item in skill_summary if str(item).strip()][:5]
    directives["focus_topics"] = [_review_item_text(item) for item in (weekly_review.get("action_items", []) or [])][:5]

    for text in texts:
        upper = text.upper()
        is_positive = any(token in upper for token in positive_markers)
        is_negative = any(token in upper for token in negative_markers)
        for session_name in session_names:
            if session_name in upper:
                target = "favor_sessions" if is_positive and not is_negative else "avoid_sessions" if is_negative else None
                if target and session_name not in directives[target]:
                    directives[target].append(session_name)
        for regime_name in regime_names:
            if regime_name in upper:
                target = "favor_regimes" if is_positive and not is_negative else "avoid_regimes" if is_negative else None
                normalized_regime = normalize_regime_label(regime_name)
                if target and normalized_regime not in directives[target]:
                    directives[target].append(normalized_regime)
    return directives


def _is_fresh(path: pathlib.Path, max_age_sec: int) -> bool:
    if not path.exists():
        return False
    return (time.time() - path.stat().st_mtime) <= max_age_sec


def _top_bucket(group: dict[str, Any], limit: int = 3) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []
    for name, stats in group.items():
        if not isinstance(stats, dict):
            continue
        rows.append(
            {
                "name": str(name),
                "trades": int(stats.get("trades", 0) or 0),
                "win_rate_pct": float(stats.get("win_rate_pct", 0.0) or 0.0),
                "avg_pnl": float(stats.get("avg_pnl", 0.0) or 0.0),
                "total_pnl": float(stats.get("total_pnl", 0.0) or 0.0),
                "profit_factor": float(stats.get("profit_factor", 0.0) or 0.0),
                "block": bool(stats.get("block", False)),
                "usable": bool(stats.get("usable", False)),
                "live_multiplier": float(stats.get("live_multiplier", 1.0) or 1.0),
                "confidence_bias": float(stats.get("confidence_bias", 0.0) or 0.0),
            }
        )
    rows.sort(key=lambda item: (item["trades"], item["win_rate_pct"], item["total_pnl"]), reverse=True)
    return rows[:limit]


def _extract_json_block(text: str) -> dict[str, Any] | None:
    if not text:
        return None
    start_idx = text.find("{")
    end_idx = text.rfind("}")
    if start_idx == -1 or end_idx == -1 or end_idx <= start_idx:
        return None
    try:
        payload = json.loads(text[start_idx : end_idx + 1])
    except Exception:
        return None
    return payload if isinstance(payload, dict) else None


def _run_hermes_agent(session_id: str, prompt: str, timeout: int = 90) -> dict[str, Any]:
    role = ROLE_RISK_MANAGER
    if "macro-context" in session_id:
        role = ROLE_TREND_ANALYST
    elif "weekly-review" in session_id:
        role = ROLE_RISK_MANAGER
    content = call_llm(role, prompt, temperature=0.1, max_tokens=1500, timeout=timeout)
    if not content:
        raise RuntimeError("LLM returned empty output")
    payload = _extract_json_block(content)
    if not payload:
        raise RuntimeError(f"Unable to parse LLM output: {content[:400]}")
    return payload


def _spawn_self(args: list[str]) -> None:
    script = pathlib.Path(__file__).resolve()
    from src.subprocess_util import popen as _sp_popen
    popen_kwargs = {
        "cwd": str(APP_DIR),
        "stdout": subprocess.DEVNULL,
        "stderr": subprocess.DEVNULL,
    }
    if sys.platform != "win32":
        popen_kwargs["start_new_session"] = True
    _sp_popen([sys.executable, str(script), *args], **popen_kwargs)


def _local_trade_snapshot() -> dict[str, Any]:
    artifact = _read_json(LOGS_DIR / "performance_factors.json")
    quality = artifact.get("quality", {}) if isinstance(artifact.get("quality", {}), dict) else {}
    snapshot = {
        "generated_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
        "source": "local",
        "quality": quality,
        "top_buckets": {
            "session": _top_bucket(artifact.get("session", {}) if isinstance(artifact.get("session", {}), dict) else {}),
            "regime": _top_bucket(artifact.get("regime", {}) if isinstance(artifact.get("regime", {}), dict) else {}),
            "strategy": _top_bucket(artifact.get("strategy", {}) if isinstance(artifact.get("strategy", {}), dict) else {}),
            "interpreter": _top_bucket(artifact.get("interpreter", {}) if isinstance(artifact.get("interpreter", {}), dict) else {}),
        },
        "summary": {
            "usable_trade_count": int(quality.get("usable_trade_count", 0) or 0),
            "excluded_incomplete_count": int(quality.get("excluded_incomplete_count", 0) or 0),
            "excluded_recovered_count": int(quality.get("excluded_recovered_count", 0) or 0),
            "ready": bool(quality.get("ready", False)),
        },
        "today_summary": artifact.get("today_summary") or {},
    }
    return snapshot


def _local_trade_brief(snapshot: dict[str, Any]) -> dict[str, Any]:
    top_session = ((snapshot.get("top_buckets", {}) or {}).get("session", []) or [{}])[0]
    top_regime = ((snapshot.get("top_buckets", {}) or {}).get("regime", []) or [{}])[0]
    weak_sessions = [
        row
        for row in ((snapshot.get("top_buckets", {}) or {}).get("session", []) or [])
        if bool(row.get("block", False)) or float(row.get("avg_pnl", 0.0) or 0.0) < 0
    ]
    today = snapshot.get("today_summary") or {}
    today_count = today.get("trade_count", 0)
    today_stats = today.get("stats")
    today_pnl = 0.0
    today_win_rate = 0.0
    if today_stats:
        today_pnl = float(today_stats.get("total_pnl", 0.0) or 0.0)
        today_win_rate = float(today_stats.get("win_rate_pct", 0.0) or 0.0)

    # Build honest narrative
    narrative = []
    if today_count > 0:
        narrative.append(f"今日({today.get('date_utc', '')}) 真实交易 {today_count} 笔，PnL {today_pnl:+.2f}，胜率 {today_win_rate:.1f}%")
    else:
        narrative.append(f"今日({today.get('date_utc', '')}) 无真实成交")
    historical = int(((snapshot.get("summary", {}) or {}).get('usable_trade_count', 0) or 0))
    narrative.append(f"历史窗口样本 {historical} 笔（用于策略参考，不代表今日表现）")
    if top_session.get("name") and top_session.get("win_rate_pct"):
        narrative.append(f"强势时段: {top_session.get('name')} 胜率{top_session.get('win_rate_pct', 0):.1f}%")

    return {
        "generated_at": snapshot.get("generated_at"),
        "source": snapshot.get("source", "local"),
        "today_summary": today,
        "highlights": {
            "best_session": top_session,
            "best_regime": top_regime,
            "caution_sessions": weak_sessions[:3],
        },
        "narrative": narrative,
    }


def refresh_trade_post_event_snapshots(force: bool = False) -> dict[str, Any]:
    if not force and _is_fresh(HERMES_TRADE_SNAPSHOT_FILE, 120):
        return _read_json(HERMES_TRADE_SNAPSHOT_FILE)

    snapshot = _local_trade_snapshot()
    brief = _local_trade_brief(snapshot)

    prompt = f"""
你是 Hermes Core 的交易知识整理器。
必须调用 `trading_factors_query` 工具优先执行 `action=summary`。
如果工具返回成功，再结合下面的本地快照补充输出。

本地快照：
{json.dumps(snapshot, ensure_ascii=False, indent=2)}

只输出 JSON：
{{
  "generated_at": "...",
  "source": "hermes",
  "best_session_regime": [],
  "failure_patterns": [],
  "regime_direction": [],
  "what_to_avoid": [],
  "memory_explainability": [],
  "summary": {{}}
}}
""".strip()
    try:
        hermes_payload = _run_hermes_agent("gold-hermes-trade-factors", prompt, timeout=45)
        if isinstance(hermes_payload, dict):
            snapshot["hermes"] = hermes_payload
            brief["hermes"] = {
                "best_session_regime": hermes_payload.get("best_session_regime", []),
                "failure_patterns": hermes_payload.get("failure_patterns", []),
                "what_to_avoid": hermes_payload.get("what_to_avoid", []),
            }
            snapshot["source"] = "local+hermes"
            brief["source"] = "local+hermes"
    except Exception as exc:
        snapshot["hermes_error"] = str(exc)
        brief["hermes_error"] = str(exc)

    _write_json(HERMES_TRADE_SNAPSHOT_FILE, snapshot)
    _write_json(HERMES_TRADE_BRIEF_FILE, brief)
    return snapshot


def schedule_trade_post_event_refresh() -> None:
    if _is_fresh(HERMES_TRADE_SNAPSHOT_FILE, 90):
        return
    _spawn_self(["trade"])


def load_trade_snapshot(max_age_sec: int = 900) -> dict[str, Any]:
    if _is_fresh(HERMES_TRADE_SNAPSHOT_FILE, max_age_sec):
        return _read_json(HERMES_TRADE_SNAPSHOT_FILE)
    return refresh_trade_post_event_snapshots(force=True)


def build_macro_context(session_data: dict[str, Any], structure_snapshot: dict[str, Any], structure_memory: dict[str, Any]) -> dict[str, Any]:
    trade_snapshot = load_trade_snapshot(max_age_sec=900)
    fallback = {
        "generated_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
        "source": "local",
        "recommended_biases": [],
        "avoid_patterns": [],
        "best_session_regime": ((trade_snapshot.get("hermes", {}) or {}).get("best_session_regime", []) or [])[:3],
        "summary": {
            "session_direction": session_data.get("overall_direction", "NEUTRAL"),
            "session_name": session_data.get("session_name", "UNKNOWN"),
            "market_structure": structure_snapshot.get("market_structure", "UNKNOWN"),
        },
    }

    prompt = f"""
你是 Hermes Core 的宏观上下文压缩器。
请优先使用 trading_factors_query 工具结合当前交易数据，输出给宏观 Agent 的短摘要。

Current session data:
{json.dumps(session_data, ensure_ascii=False, indent=2)}

Current structure snapshot:
{json.dumps(structure_snapshot, ensure_ascii=False, indent=2)}

Recent structure memory:
{json.dumps(structure_memory, ensure_ascii=False, indent=2)}

Trade factor snapshot:
{json.dumps(trade_snapshot, ensure_ascii=False, indent=2)}

只输出 JSON：
{{
  "generated_at": "...",
  "source": "hermes",
  "recommended_biases": [],
  "avoid_patterns": [],
  "best_session_regime": [],
  "summary": {{
    "macro_bias": "LONG|SHORT|NEUTRAL",
    "focus": "",
    "caution": ""
  }}
}}
""".strip()
    try:
        payload = _run_hermes_agent("gold-hermes-macro-context", prompt, timeout=90)
        if isinstance(payload, dict):
            fallback = payload
            fallback.setdefault("source", "hermes")
    except Exception as exc:
        fallback["hermes_error"] = str(exc)

    _write_json(HERMES_MACRO_CONTEXT_FILE, fallback)
    return fallback


def load_macro_context() -> dict[str, Any]:
    return _read_json(HERMES_MACRO_CONTEXT_FILE)


def _local_governance_brief(kind: str, payload: dict[str, Any]) -> dict[str, Any]:
    if kind == "approval_chain":
        decision = payload.get("decision", {}) if isinstance(payload.get("decision", {}), dict) else {}
        return {
            "generated_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
            "kind": kind,
            "source": "local",
            "status": str(decision.get("status", "UNKNOWN") or "UNKNOWN"),
            "next_step": str(decision.get("nextStep", "") or ""),
            "missing_conditions": decision.get("missingConditions", []),
            "recommended_actions": decision.get("recommendedActions", []),
        }
    training_lifecycle = payload.get("trainingLifecycle", {}) if isinstance(payload.get("trainingLifecycle", {}), dict) else {}
    auto_governance = payload.get("autoGovernance", {}) if isinstance(payload.get("autoGovernance", {}), dict) else {}
    return {
        "generated_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
        "kind": kind,
        "source": "local",
        "training_status": str("READY" if training_lifecycle.get("readyForTraining", False) else "BLOCKED"),
        "blocked_reasons": training_lifecycle.get("blockedReasons", []),
        "auto_governance_status": str(auto_governance.get("status", "NOOP") or "NOOP"),
        "auto_governance_reason": str(auto_governance.get("lastReason", "") or ""),
    }


def _compact_governance_payload(kind: str, payload: dict[str, Any]) -> dict[str, Any]:
    if kind == "approval_chain":
        decision = payload.get("decision", {}) if isinstance(payload.get("decision", {}), dict) else {}
        quality_context = decision.get("qualityContext", {}) if isinstance(decision.get("qualityContext", {}), dict) else {}
        shadow_learner = payload.get("shadowLearner", {}) if isinstance(payload.get("shadowLearner", {}), dict) else {}
        live_learner = payload.get("liveLearner", {}) if isinstance(payload.get("liveLearner", {}), dict) else {}
        auto_governance = payload.get("autoGovernance", {}) if isinstance(payload.get("autoGovernance", {}), dict) else {}
        phase_readiness = payload.get("phaseReadiness", {}) if isinstance(payload.get("phaseReadiness", {}), dict) else {}
        return {
            "stage": payload.get("stage", "UNKNOWN"),
            "decision": {
                "status": decision.get("status", "UNKNOWN"),
                "next_step": decision.get("nextStep", ""),
                "can_promote": decision.get("canPromote", False),
                "review_pending": decision.get("reviewPending", False),
                "promotion_mode": decision.get("promotionMode", ""),
                "promotion_target": decision.get("promotionTarget", ""),
                "missing_conditions": decision.get("missingConditions", []),
                "recommended_actions": decision.get("recommendedActions", []),
            },
            "quality_context": {
                "ready_for_training": quality_context.get("ready_for_training", False),
                "blocked_reasons": quality_context.get("blocked_reasons", []),
                "closure_coverage_pct": quality_context.get("closure_coverage_pct", 0.0),
                "critical_anomalies": quality_context.get("critical_anomalies", 0),
                "recovered_closed_ratio_pct": quality_context.get("recovered_closed_ratio_pct", 0.0),
            },
            "shadow_learner": {
                "status": shadow_learner.get("status", "EMPTY"),
                "divergence_count": shadow_learner.get("divergenceCount", 0),
            },
            "live_learner": {
                "status": live_learner.get("status", "EMPTY"),
                "learner_live_rate": live_learner.get("learnerLiveRate", 0.0),
            },
            "auto_governance": {
                "status": auto_governance.get("status", "NOOP"),
                "last_reason": auto_governance.get("lastReason", ""),
            },
            "phase_readiness": {
                "status": phase_readiness.get("status", "BLOCKED"),
            },
        }

    training_lifecycle = payload.get("trainingLifecycle", {}) if isinstance(payload.get("trainingLifecycle", {}), dict) else {}
    closure_lifecycle = payload.get("closureLifecycle", {}) if isinstance(payload.get("closureLifecycle", {}), dict) else {}
    shadow_learner = payload.get("shadowLearner", {}) if isinstance(payload.get("shadowLearner", {}), dict) else {}
    live_learner = payload.get("liveLearner", {}) if isinstance(payload.get("liveLearner", {}), dict) else {}
    auto_governance = payload.get("autoGovernance", {}) if isinstance(payload.get("autoGovernance", {}), dict) else {}
    rollback_evidence = payload.get("rollbackEvidence", {}) if isinstance(payload.get("rollbackEvidence", {}), dict) else {}
    phase_readiness = payload.get("phaseReadiness", {}) if isinstance(payload.get("phaseReadiness", {}), dict) else {}
    return {
        "training_lifecycle": {
            "ready_for_training": training_lifecycle.get("readyForTraining", False),
            "blocked_reasons": training_lifecycle.get("blockedReasons", []),
            "closure_coverage_pct": training_lifecycle.get("closureCoveragePct", 0.0),
            "critical_anomalies": training_lifecycle.get("criticalAnomalies", 0),
            "recovered_closed_ratio_pct": training_lifecycle.get("recoveredClosedRatioPct", 0.0),
        },
        "closure_lifecycle": {
            "normal_closed_count": closure_lifecycle.get("normalClosedCount", 0),
            "recovered_closed_count": closure_lifecycle.get("recoveredClosedCount", 0),
            "pending_repair_count": closure_lifecycle.get("pendingRepairCount", 0),
        },
        "shadow_learner": {
            "status": shadow_learner.get("status", "EMPTY"),
            "total": shadow_learner.get("total", 0),
            "action_match_rate": shadow_learner.get("actionMatchRate", 0.0),
            "direction_match_rate": shadow_learner.get("directionMatchRate", 0.0),
        },
        "live_learner": {
            "status": live_learner.get("status", "EMPTY"),
            "current_mode": live_learner.get("currentMode", "shadow"),
            "learner_live_rate": live_learner.get("learnerLiveRate", 0.0),
        },
        "auto_governance": {
            "status": auto_governance.get("status", "NOOP"),
            "last_reason": auto_governance.get("lastReason", ""),
        },
        "rollback_evidence": {
            "triggered": rollback_evidence.get("triggered", []),
            "kill_switch": rollback_evidence.get("killSwitch", False),
        },
        "phase_readiness": {
            "status": phase_readiness.get("status", "BLOCKED"),
        },
    }


def refresh_governance_brief(kind: str, payload: dict[str, Any]) -> dict[str, Any]:
    local_brief = _local_governance_brief(kind, payload)
    trade_snapshot = _read_json(HERMES_TRADE_SNAPSHOT_FILE)
    compact_payload = _compact_governance_payload(kind, payload)
    prompt = f"""
你是 Hermes Core 的训练与治理摘要助手。
请结合当前治理数据和交易因子快照，输出简洁 JSON 摘要。

kind: {kind}
payload:
{json.dumps(compact_payload, ensure_ascii=False, indent=2)}

trade_snapshot:
{json.dumps(trade_snapshot, ensure_ascii=False, indent=2)}

只输出 JSON：
{{
  "generated_at": "...",
  "kind": "{kind}",
  "source": "hermes",
  "summary": [],
  "risks": [],
  "historical_hints": [],
  "next_step": ""
}}
""".strip()
    try:
        hermes_payload = _run_hermes_agent(f"gold-hermes-{kind}", prompt, timeout=45)
        if isinstance(hermes_payload, dict):
            filtered_risks = hermes_payload.get("risks", []) if isinstance(hermes_payload.get("risks", []), list) else []
            if kind == "approval_chain":
                blocked_reasons = [str(item) for item in local_brief.get("blocked_reasons", []) if str(item).strip()]
                quality_blocked = any("quality" in reason.lower() or "anomal" in reason.lower() for reason in blocked_reasons)
                if not quality_blocked:
                    filtered_risks = [
                        item
                        for item in filtered_risks
                        if "quality gate" not in str(item.get("item", "")).lower()
                        and "critical_anomal" not in str(item.get("detail", "")).lower()
                    ]
            local_brief["hermes"] = {
                "historical_hints": hermes_payload.get("historical_hints", []),
                "risks": filtered_risks,
            }
            local_brief["source"] = "local+hermes"
    except Exception as exc:
        local_brief["hermes_error"] = str(exc)
    target_file = HERMES_TRAINING_BRIEF_FILE if kind == "approval_chain" else HERMES_GOVERNANCE_BRIEF_FILE
    _write_json(target_file, local_brief)
    return local_brief


def schedule_governance_brief_refresh(kind: str, payload: dict[str, Any]) -> None:
    payload_file = PROJECT_STATE_DIR / f"hermes_{kind}_payload.json"
    _write_json(payload_file, payload)
    _spawn_self(["governance", kind, str(payload_file)])


def load_governance_brief(kind: str) -> dict[str, Any]:
    target_file = HERMES_TRAINING_BRIEF_FILE if kind == "approval_chain" else HERMES_GOVERNANCE_BRIEF_FILE
    return _read_json(target_file)


def load_system_design_contract() -> dict[str, Any]:
    return _read_json(SYSTEM_DESIGN_CONTRACT_FILE)


def _derive_runtime_rules(
    trade_snapshot: dict[str, Any],
    weekly_review: dict[str, Any],
    performance_factors: dict[str, Any],
) -> dict[str, Any]:
    session_rules: dict[str, dict[str, Any]] = {}
    regime_rules: dict[str, dict[str, Any]] = {}
    governance_priority: list[dict[str, Any]] = []
    review_directives = _extract_review_directives(weekly_review)

    session_group = performance_factors.get("session", {}) if isinstance(performance_factors.get("session", {}), dict) else {}
    for session_name, stats in session_group.items():
        if not isinstance(stats, dict):
            continue
        session_key = str(session_name).upper()
        win_rate = float(stats.get("win_rate_pct", 0.0) or 0.0)
        pf = float(stats.get("profit_factor", 0.0) or 0.0)
        recent_loss_streak = int(stats.get("recent_loss_streak", 0) or 0)
        sample_size = int(
            stats.get("sample_size")
            or stats.get("trade_count")
            or stats.get("trades")
            or stats.get("count")
            or 0
        )
        live_multiplier = float(stats.get("live_multiplier", 1.0) or 1.0)
        confidence_bias = float(stats.get("confidence_bias", 0.0) or 0.0)
        action = "neutral"
        limit_only = False
        confidence_delta = 0.0
        size_multiplier = 1.0
        reason_parts: list[str] = []
        evidence_reliable = sample_size >= 40
        if win_rate >= 75.0 and pf >= 1.5:
            action = "favor"
            confidence_delta = min(0.04, max(0.0, confidence_bias))
            size_multiplier = max(1.0, live_multiplier)
            reason_parts.append(f"win_rate={win_rate:.1f}")
            reason_parts.append(f"pf={pf:.2f}")
        elif pf < 0.8 or recent_loss_streak >= 2:
            action = "avoid"
            limit_only = True
            confidence_delta = -0.04 if pf < 0.8 else -0.02
            size_multiplier = min(0.88, live_multiplier)
            reason_parts.append(f"pf={pf:.2f}")
            if recent_loss_streak:
                reason_parts.append(f"loss_streak={recent_loss_streak}")
        if str(session_name).upper() in review_directives.get("favor_sessions", []):
            action = "favor"
            confidence_delta = max(confidence_delta, 0.03)
            size_multiplier = max(size_multiplier, 1.02)
            reason_parts.append("weekly_review_favor")
        if str(session_name).upper() in review_directives.get("avoid_sessions", []):
            action = "avoid"
            limit_only = True
            confidence_delta = min(confidence_delta, -0.03)
            size_multiplier = min(size_multiplier, 0.9)
            reason_parts.append("weekly_review_avoid")
        # 亚欧重叠是交接窗口：默认中性偏机会，但比欧美重叠更保守，要求结构确认。
        if session_key == "ASIA_LONDON_OVERLAP" and "weekly_review_avoid" not in reason_parts:
            if not evidence_reliable:
                action = "neutral"
                limit_only = False
                confidence_delta = max(confidence_delta, 0.0)
                size_multiplier = max(0.98, min(size_multiplier, 1.02))
                reason_parts.append("asia_london_overlap_cautious_default_low_evidence")
            elif pf >= 0.78 or (win_rate >= 50.0 and recent_loss_streak <= 1):
                action = "favor"
                limit_only = False
                confidence_delta = max(confidence_delta, 0.01)
                size_multiplier = max(size_multiplier, 1.0)
                reason_parts.append("asia_london_overlap_cautious_opportunity")
            elif pf <= 0.55 and recent_loss_streak >= 2 and win_rate < 45.0:
                action = "avoid"
                limit_only = True
                confidence_delta = min(confidence_delta, -0.03)
                size_multiplier = min(size_multiplier, 0.9)
                reason_parts.append("asia_london_overlap_evidence_deteriorated")

        # 欧美重叠是高波动机会时段：默认机会优先，只有“证据充分且显著恶化”才转强防守。
        if session_key == "LONDON_NY_OVERLAP" and "weekly_review_avoid" not in reason_parts:
            if not evidence_reliable:
                action = "favor"
                limit_only = False
                confidence_delta = max(confidence_delta, 0.01)
                size_multiplier = max(size_multiplier, 1.0)
                reason_parts.append("overlap_opportunity_default_low_evidence")
            elif pf >= 0.72 or (win_rate >= 48.0 and recent_loss_streak <= 1):
                action = "favor"
                limit_only = False
                confidence_delta = max(confidence_delta, 0.01)
                size_multiplier = max(size_multiplier, 1.0)
                reason_parts.append("overlap_opportunity_default")
            elif pf <= 0.50 and recent_loss_streak >= 3 and win_rate < 42.0:
                action = "avoid"
                limit_only = True
                confidence_delta = min(confidence_delta, -0.03)
                size_multiplier = min(size_multiplier, 0.9)
                reason_parts.append("overlap_evidence_deteriorated")

        hard_block = evidence_reliable and (
            pf <= 0.50 or (recent_loss_streak >= 4 and win_rate < 40.0)
        )
        if session_key == "LONDON_NY_OVERLAP":
            hard_block = (
                sample_size >= 60
                and pf <= 0.45
                and recent_loss_streak >= 4
                and win_rate < 40.0
            )
        elif session_key == "ASIA_LONDON_OVERLAP":
            hard_block = (
                sample_size >= 50
                and pf <= 0.50
                and recent_loss_streak >= 3
                and win_rate < 42.0
            )
        if hard_block:
            decision_matrix = {
                "high_conf": {
                    "min_confidence": 0.82,
                    "mode": "LIMIT",
                    "require_confirmation": True,
                    "require_pullback": ["HEALTHY", "STRONG"],
                    "max_rsi_buy": 78,
                    "min_rsi_sell": 22,
                },
                "mid_conf": {
                    "min_confidence": 0.40,
                    "mode": "REVIEW",
                    "require_confirmation": False,
                    "require_pullback": [],
                },
                "low_conf": {
                    "mode": "BLOCK",
                },
            }
        elif action == "avoid":
            decision_matrix = {
                "high_conf": {
                    "min_confidence": 0.80,
                    "mode": "ALLOW_MARKET_IF_TREND",
                    "require_confirmation": True,
                    "require_pullback": ["HEALTHY", "STRONG"],
                    "min_adx": 24,
                    "max_rsi_buy": 80,
                    "min_rsi_sell": 20,
                },
                "mid_conf": {
                    "min_confidence": 0.68,
                    "mode": "LIMIT",
                    "require_confirmation": False,
                    "require_pullback": [],
                },
                "low_conf": {
                    "mode": "REVIEW",
                },
            }
        elif action == "favor":
            decision_matrix = {
                "high_conf": {
                    "min_confidence": 0.76,
                    "mode": "ALLOW_MARKET_IF_TREND",
                    "require_confirmation": False,
                    "require_pullback": [],
                    "min_adx": 20,
                },
                "mid_conf": {
                    "min_confidence": 0.64,
                    "mode": "LIMIT",
                    "require_confirmation": False,
                    "require_pullback": [],
                },
                "low_conf": {
                    "mode": "REVIEW",
                },
            }
        elif session_key == "LONDON_NY_OVERLAP":
            decision_matrix = {
                "high_conf": {
                    "min_confidence": 0.76,
                    "mode": "ALLOW_MARKET_IF_TREND",
                    "require_confirmation": False,
                    "require_pullback": [],
                    "min_adx": 22,
                    "max_rsi_buy": 82,
                    "min_rsi_sell": 18,
                },
                "mid_conf": {
                    "min_confidence": 0.64,
                    "mode": "LIMIT",
                    "require_confirmation": False,
                    "require_pullback": [],
                },
                "low_conf": {
                    "mode": "REVIEW",
                },
            }
        elif session_key == "ASIA_LONDON_OVERLAP":
            decision_matrix = {
                "high_conf": {
                    "min_confidence": 0.78,
                    "mode": "ALLOW_MARKET_IF_TREND",
                    "require_confirmation": True,
                    "require_pullback": ["HEALTHY", "STRONG"],
                    "min_adx": 22,
                    "max_rsi_buy": 80,
                    "min_rsi_sell": 20,
                },
                "mid_conf": {
                    "min_confidence": 0.66,
                    "mode": "LIMIT",
                    "require_confirmation": False,
                    "require_pullback": [],
                },
                "low_conf": {
                    "mode": "REVIEW",
                },
            }
        else:
            decision_matrix = {
                "high_conf": {
                    "min_confidence": 0.80,
                    "mode": "LIMIT",
                    "require_confirmation": True,
                    "require_pullback": ["HEALTHY", "STRONG"],
                },
                "mid_conf": {
                    "min_confidence": 0.66,
                    "mode": "LIMIT",
                    "require_confirmation": False,
                    "require_pullback": [],
                },
                "low_conf": {
                    "mode": "REVIEW",
                },
            }
        session_rules[str(session_name).upper()] = {
            "action": action,
            "limit_only": limit_only,
            "block": hard_block,
            "confidence_delta": round(confidence_delta, 4),
            "size_multiplier": round(size_multiplier, 4),
            "decision_matrix": decision_matrix,
            "sample_size": sample_size,
            "evidence_reliable": evidence_reliable,
            "reason": " | ".join(reason_parts) if reason_parts else "derived_from_performance",
        }

    if "ASIA_LONDON_OVERLAP" not in session_rules:
        london_ref = session_rules.get("LONDON", {}) if isinstance(session_rules.get("LONDON", {}), dict) else {}
        inherited_matrix = (
            london_ref.get("decision_matrix", {})
            if isinstance(london_ref.get("decision_matrix", {}), dict)
            else {}
        )
        session_rules["ASIA_LONDON_OVERLAP"] = {
            "action": "neutral",
            "limit_only": False,
            "block": False,
            "confidence_delta": 0.0,
            "size_multiplier": 1.0,
            "decision_matrix": inherited_matrix
            or {
                "high_conf": {
                    "min_confidence": 0.78,
                    "mode": "ALLOW_MARKET_IF_TREND",
                    "require_confirmation": True,
                    "require_pullback": ["HEALTHY", "STRONG"],
                    "min_adx": 22,
                },
                "mid_conf": {
                    "min_confidence": 0.66,
                    "mode": "LIMIT",
                    "require_confirmation": False,
                    "require_pullback": [],
                },
                "low_conf": {"mode": "REVIEW"},
            },
            "sample_size": 0,
            "evidence_reliable": False,
            "reason": "asia_london_overlap_default_from_london",
        }

    regime_group = performance_factors.get("regime", {}) if isinstance(performance_factors.get("regime", {}), dict) else {}
    for regime_name, stats in regime_group.items():
        if not isinstance(stats, dict):
            continue
        regime_key = str(regime_name).upper()
        pf = float(stats.get("profit_factor", 0.0) or 0.0)
        avg_pnl = float(stats.get("avg_pnl", 0.0) or 0.0)
        live_multiplier = float(stats.get("live_multiplier", 1.0) or 1.0)
        confidence_bias = float(stats.get("confidence_bias", 0.0) or 0.0)
        action = "neutral"
        limit_only = False
        confidence_floor_delta = 0.0
        size_multiplier = 1.0
        if pf >= 1.3 and avg_pnl > 0:
            action = "favor"
            confidence_floor_delta = min(0.03, max(0.0, confidence_bias))
            size_multiplier = max(1.0, live_multiplier)
        elif pf < 0.85 and avg_pnl <= 0:
            action = "avoid"
            limit_only = True
            confidence_floor_delta = -0.03
            size_multiplier = min(0.9, live_multiplier)
        regime_key = normalize_regime_label(regime_key)
        favor_regimes = {normalize_regime_label(item) for item in review_directives.get("favor_regimes", [])}
        avoid_regimes = {normalize_regime_label(item) for item in review_directives.get("avoid_regimes", [])}
        if regime_key in favor_regimes:
            action = "favor"
            confidence_floor_delta = max(confidence_floor_delta, 0.02)
            size_multiplier = max(size_multiplier, 1.02)
        if regime_key in avoid_regimes:
            action = "avoid"
            limit_only = True
            confidence_floor_delta = min(confidence_floor_delta, -0.03)
            size_multiplier = min(size_multiplier, 0.9)
        # TRANSITION keeps strong defense, but must not hard-freeze the system.
        if regime_key == "TRANSITION" and action == "avoid":
            limit_only = True
            confidence_floor_delta = min(confidence_floor_delta, -0.02)
            size_multiplier = min(size_multiplier, 0.82)
        hard_block = regime_key != "TRANSITION" and pf <= 0.60 and avg_pnl < 0
        regime_rules[regime_key] = {
            "action": action,
            "limit_only": limit_only,
            "block": hard_block,
            "high_confidence_exception": 0.80 if regime_key == "TRANSITION" else None,
            "confidence_floor_delta": round(confidence_floor_delta, 4),
            "size_multiplier": round(size_multiplier, 4),
            "reason": f"pf={pf:.2f} | avg_pnl={avg_pnl:.2f}",
        }

    training_status = _read_json(DATA_DIR / "offline" / "training_status.json")
    quality_gate = training_status.get("qualityGate", {}) if isinstance(training_status.get("qualityGate", {}), dict) else {}
    blocked_reasons = quality_gate.get("blocked_reasons", []) if isinstance(quality_gate.get("blocked_reasons", []), list) else []
    perf_quality = performance_factors.get("quality", {}) if isinstance(performance_factors.get("quality", {}), dict) else {}
    usable_trade_count = int(perf_quality.get("usable_trade_count", 0) or 0)
    governance_priority.append(
        {
            "type": "training_ready",
            "count": 1 if bool(quality_gate.get("ready_for_training", False)) else 0,
            "message": (
                "training gate ready"
                if bool(quality_gate.get("ready_for_training", False))
                else ", ".join(str(item) for item in blocked_reasons) or "training gate blocked"
            ),
        }
    )
    governance_priority.append(
        {
            "type": "critical_anomalies",
            "count": int(quality_gate.get("critical_anomalies", 0) or 0),
            "message": f"critical={int(quality_gate.get('critical_anomalies', 0) or 0)}",
        }
    )
    governance_priority.append(
        {
            "type": "usable_trade_count",
            "count": usable_trade_count,
            "message": f"usable={usable_trade_count}",
        }
    )
    promotion_mode = "shadow"
    next_mode = "paper"
    try:
        from src.rollout_manager import get_promotion_mode, evaluate_promotion_gates

        promotion_mode = get_promotion_mode(APP_DIR, "default")
        # Pass None to trigger load_paper_trading_stats() fallback (was {} which skips fallback)
        gates = evaluate_promotion_gates(APP_DIR, "default", None)
        pending = [item for item in gates if not bool(item.get("ready", False))]
        ready = [item for item in gates if bool(item.get("ready", False))]
        next_mode = pending[0].get("next_mode", next_mode) if pending else (ready[0].get("next_mode", next_mode) if ready else next_mode)
        governance_priority.append(
            {
                "type": "promotion_pending",
                "count": len(pending),
                "message": (
                    f"{promotion_mode}->{next_mode}: " + ", ".join(f"{item.get('gate')}={item.get('current')}/{item.get('required')}" for item in pending[:3])
                    if pending
                    else f"{promotion_mode}->{next_mode}: all gates ready"
                ),
            }
        )

        # ── Write governance decision (fixes Break 2: auto_governance write path) ──
        if ready and next_mode:
            try:
                import json as _json
                _gov_file = LOGS_DIR / "auto_governance_status_latest.json"
                _gov_payload = {
                    "status": "PROMOTION_READY" if len(ready) >= len(gates) else "PARTIAL_GATES_PASSED",
                    "current_mode": promotion_mode,
                    "next_mode": next_mode,
                    "last_action": f"promote_to_{next_mode}" if len(ready) >= len(gates) else "gates_partial",
                    "gates_summary": {
                        "total": len(gates), "ready": len(ready), "pending": len(pending),
                        "ready_items": [{"gate": r["gate"], "current": r["current"], "required": r["required"]} for r in ready],
                    },
                    "updated_at": datetime.now(timezone.utc).isoformat(),
                    "source": "hermes_bridge",
                }
                _gov_file.parent.mkdir(parents=True, exist_ok=True)
                _gov_file.write_text(_json.dumps(_gov_payload, indent=2, ensure_ascii=False), encoding="utf-8")
            except Exception:
                pass
        # ── End governance write ──
    except Exception:
        pass

    # ── semi-auto injection of low-risk rules (safety-gated) ──
    # Only confidence_delta and size_multiplier from high-confidence sessions
    # are auto-injected.  BLOCK / LIMIT / mode changes always stay manual.
    # Controlled by:
    #   1. HERMES_AUTO_INJECT env var (explicit override)
    #   2. strategy_rules_v3.json → hermes.auto_inject (config-driven)
    import os as _os
    _auto_inject = _os.environ.get("HERMES_AUTO_INJECT", "").lower() in ("1", "true", "yes")
    # Check config-driven toggle if env var not explicitly set
    if not _auto_inject and not _os.environ.get("HERMES_AUTO_INJECT"):
        try:
            import config_manager as _cm
            rules = _cm.load_strategy_rules()
            hermes_cfg = rules.get("hermes", {}) if isinstance(rules, dict) else {}
            _auto_inject = bool(hermes_cfg.get("auto_inject", True))  # default ON
        except Exception:
            _auto_inject = True  # Safe default: enable low-risk auto-injection
    # Safety thresholds — configurable via strategy_rules_v3.json → hermes
    _hermes_safety = {
        "min_sample_size": 60,
        "min_win_rate": 70,
        "max_confidence_delta": 0.20,
        "max_size_delta": 0.25,
    }
    try:
        import config_manager as _cm2
        rules = _cm2.load_strategy_rules()
        hermes_cfg = rules.get("hermes", {}) if isinstance(rules, dict) else {}
        if isinstance(hermes_cfg.get("safety"), dict):
            _hermes_safety.update(hermes_cfg["safety"])
    except Exception:
        pass
    safe_session_rules: dict[str, Any] = {}
    if _auto_inject:
        for sess_name, features in (session_rules or {}).items():
            sample_size = int(features.get("sample_size", 0) or 0)
            win_rate = float(features.get("win_rate", 0) or 0)
            conf_delta = abs(float(features.get("confidence_delta", 0) or 0))
            size_delta = abs(float(features.get("size_multiplier", 1.0) or 1.0) - 1.0)
            if (sample_size >= _hermes_safety["min_sample_size"] and
                win_rate >= _hermes_safety["min_win_rate"] and
                conf_delta <= _hermes_safety["max_confidence_delta"] and
                size_delta <= _hermes_safety["max_size_delta"]):
                safe_session_rules[sess_name] = {
                    "confidence_delta": features.get("confidence_delta", 0),
                    "size_multiplier": features.get("size_multiplier", 1.0),
                    "auto_injected": True,
                }

    return {
        "generated_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
        "source": "derived",
        "design_contract": load_system_design_contract(),
        # Hermes 仅提供离线学习/进化特征，不作为在线策略规则入口。
        "session_offline_features": session_rules,
        "regime_offline_features": regime_rules,
        # 兼容字段：低风险规则可在安全闸下自动注入，高风险规则保持空
        "session_rules": safe_session_rules if _auto_inject else {},
        "regime_rules": {},
        "policy_mode": "semi_auto_low_risk" if _auto_inject else "offline_features_only",
        "review_directives": review_directives,
        "governance_priority": governance_priority[:8],
    }


def refresh_runtime_rules(force: bool = False) -> dict[str, Any]:
    if not force and _is_fresh(HERMES_RUNTIME_RULES_FILE, 300):
        return _read_json(HERMES_RUNTIME_RULES_FILE)
    trade_snapshot = load_trade_snapshot(max_age_sec=1800)
    weekly_review = _read_json(HERMES_WEEKLY_REVIEW_JSON_FILE)
    performance_factors = _read_json(LOGS_DIR / "performance_factors.json")
    rules = _derive_runtime_rules(trade_snapshot, weekly_review, performance_factors)
    _write_json(HERMES_RUNTIME_RULES_FILE, rules)
    return rules


def load_runtime_rules(force_refresh: bool = False) -> dict[str, Any]:
    if force_refresh:
        return refresh_runtime_rules(force=True)
    return _read_json(HERMES_RUNTIME_RULES_FILE) if HERMES_RUNTIME_RULES_FILE.exists() else refresh_runtime_rules(force=True)


def _local_weekly_review() -> dict[str, Any]:
    trade_snapshot = load_trade_snapshot(max_age_sec=3600)
    macro_context = _read_json(HERMES_MACRO_CONTEXT_FILE)
    training_brief = _read_json(HERMES_TRAINING_BRIEF_FILE)
    governance_brief = _read_json(HERMES_GOVERNANCE_BRIEF_FILE)
    recent_notices = _read_recent_json("review_notify_*.json", limit=5)
    quality = trade_snapshot.get("summary", {}) if isinstance(trade_snapshot.get("summary", {}), dict) else {}
    top_sessions = ((trade_snapshot.get("top_buckets", {}) or {}).get("session", []) or [])[:3]
    top_regimes = ((trade_snapshot.get("top_buckets", {}) or {}).get("regime", []) or [])[:3]
    action_items = [
        "优先修复 incomplete trade/closure 管道，扩大 Hermes 可学习样本。",
        "继续把高频失败模式沉淀进 MEMORY 和 Skill，而不是停留在日志。",
        "周末复盘优先审视 session/regime/strategy 三类桶的胜率和利润因子漂移。",
    ]
    if int(quality.get("usable_trade_count", 0) or 0) <= 0:
        action_items.insert(0, "当前 usable_trade_count=0，先处理自然闭环样本不足。")
    return {
        "generated_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
        "source": "local",
        "title": "Hermes Weekly Review",
        "executive_summary": [
            f"当前自然闭环可用样本 {int(quality.get('usable_trade_count', 0) or 0)} 笔，incomplete 排除 {int(quality.get('excluded_incomplete_count', 0) or 0)} 笔。",
            f"宏观偏向 {((macro_context.get('summary', {}) or {}).get('macro_bias', 'NEUTRAL'))}，训练状态 {training_brief.get('status', training_brief.get('training_status', 'UNKNOWN'))}。",
            f"治理状态 {governance_brief.get('training_status', governance_brief.get('auto_governance_status', 'UNKNOWN'))}，自动治理 {governance_brief.get('auto_governance_status', 'NOOP')}。",
        ],
        "best_patterns": top_sessions + top_regimes,
        "failure_patterns": (
            ((trade_snapshot.get("hermes", {}) or {}).get("failure_patterns", []) or [])
            or recent_notices
        )[:5],
        "governance_notes": [
            training_brief.get("next_step", ""),
            ((training_brief.get("hermes", {}) or {}).get("next_step", "")),
            ((governance_brief.get("hermes", {}) or {}).get("next_step", "")),
        ],
        "action_items": [item for item in action_items if item],
        "memory_notes": [
            "优先沉淀 failure_patterns、best_session_regime、治理阻断原因。",
            "保持 Hermes 只做异步认知层，不进入实时开仓权。",
        ],
        "skill": {
            "name": HERMES_SKILL_NAME,
            "status": "pending",
            "path": str(HERMES_SKILL_PATH),
            "summary": [
                "固定周末复盘流程：读取 trade/macro/training/governance 四类快照。",
                "输出周报并沉淀下一轮复盘 Skill。",
            ],
        },
    }


def _build_skill_content(review: dict[str, Any]) -> str:
    action_items = review.get("action_items", []) if isinstance(review.get("action_items", []), list) else []
    summary = review.get("executive_summary", []) if isinstance(review.get("executive_summary", []), list) else []
    return (
        "# Hermes weekly review workflow for GoldAutoTrade\n\n"
        "## TRIGGERS\n"
        "- 周末固定复盘窗口\n"
        "- 训练治理链出现持续阻断时\n"
        "- 交易后样本质量明显恶化时\n\n"
        "## INPUTS\n"
        "- `logs/hermes_factor_snapshot.json`\n"
        "- `logs/hermes_macro_context.json`\n"
        "- `logs/hermes_training_brief.json`\n"
        "- `logs/hermes_governance_brief.json`\n\n"
        "## WORKFLOW\n"
        "- 汇总交易因子、宏观上下文、训练摘要、治理摘要。\n"
        "- 提炼 best patterns、failure patterns、governance blockers。\n"
        "- 输出周报并给出下一周优先行动。\n\n"
        "## CURRENT SUMMARY\n"
        + "\n".join(f"- {item}" for item in summary[:5])
        + "\n\n## ACTION ITEMS\n"
        + "\n".join(f"- {item}" for item in action_items[:5])
        + "\n\n## GUARDRAILS\n"
        "- 不直接参与实时开仓决策。\n"
        "- 不绕过 risk_manager。\n"
        "- 不直接写 live 参数或 MT5 指令。\n"
    )


def _ensure_weekly_skill(review: dict[str, Any]) -> dict[str, Any]:
    skill_info = review.get("skill", {}) if isinstance(review.get("skill", {}), dict) else {}
    skill_path = pathlib.Path(str(skill_info.get("path") or HERMES_SKILL_PATH))
    if skill_path.exists():
        return {
            "name": skill_info.get("name", HERMES_SKILL_NAME),
            "status": "existing",
            "path": str(skill_path),
            "summary": skill_info.get("summary", []),
        }

    content = _build_skill_content(review)
    frontmatter = (
        "---\n"
        f"name: {json.dumps(str(skill_info.get('name', HERMES_SKILL_NAME)))}\n"
        f"description: {json.dumps('Hermes weekly review workflow for GoldAutoTrade')}\n"
        f"version: {json.dumps('1.0.0')}\n"
        f"created_at: {json.dumps(time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime()))}\n"
        f"updated_at: {json.dumps(time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime()))}\n"
        "---\n\n"
    )
    _write_text(skill_path, frontmatter + content)
    return {
        "name": skill_info.get("name", HERMES_SKILL_NAME),
        "status": "local_fallback_created",
        "path": str(skill_path),
        "summary": skill_info.get("summary", []),
    }


def _render_weekly_review_markdown(review: dict[str, Any]) -> str:
    def _render_list(items: Any) -> str:
        if not isinstance(items, list) or not items:
            return "- 无\n"
        rows: list[str] = []
        for item in items:
            if isinstance(item, dict):
                title = (
                    item.get("task")
                    or item.get("name")
                    or item.get("item")
                    or item.get("pattern")
                    or item.get("type")
                    or item.get("_file")
                    or "item"
                )
                detail = (
                    item.get("detail")
                    or item.get("summary")
                    or item.get("note")
                    or item.get("message")
                    or item.get("evidence")
                    or item.get("blocker")
                    or ""
                )
                prefix = f"[{item.get('priority')}]" if item.get("priority") else ""
                body = f"{title}: {detail}".rstrip(": ")
                rows.append(f"- {prefix} {body}".replace("-  ", "- "))
            else:
                rows.append(f"- {item}")
        return "\n".join(rows) + "\n"

    skill = review.get("skill", {}) if isinstance(review.get("skill", {}), dict) else {}
    return (
        f"# {review.get('title', 'Hermes Weekly Review')}\n\n"
        f"- Generated At: {review.get('generated_at', '')}\n"
        f"- Source: {review.get('source', 'local')}\n"
        f"- Skill: {skill.get('name', HERMES_SKILL_NAME)} ({skill.get('status', 'pending')})\n"
        f"- Skill Path: {skill.get('path', str(HERMES_SKILL_PATH))}\n\n"
        "## Executive Summary\n"
        f"{_render_list(review.get('executive_summary', []))}\n"
        "## Best Patterns\n"
        f"{_render_list(review.get('best_patterns', []))}\n"
        "## Failure Patterns\n"
        f"{_render_list(review.get('failure_patterns', []))}\n"
        "## Governance Notes\n"
        f"{_render_list(review.get('governance_notes', []))}\n"
        "## Action Items\n"
        f"{_render_list(review.get('action_items', []))}\n"
        "## Memory Notes\n"
        f"{_render_list(review.get('memory_notes', []))}\n"
        "## Skill Summary\n"
        f"{_render_list(skill.get('summary', []))}"
    )


def _compact_weekly_review_payload(review: dict[str, Any]) -> dict[str, Any]:
    def _compact_item(item: Any) -> Any:
        if isinstance(item, dict):
            return {
                key: item.get(key)
                for key in ("name", "item", "pattern", "detail", "message", "timestamp", "_file")
                if item.get(key) not in (None, "")
            }
        return item

    return {
        "generated_at": review.get("generated_at"),
        "title": review.get("title"),
        "executive_summary": (review.get("executive_summary", []) or [])[:3],
        "best_patterns": [_compact_item(item) for item in (review.get("best_patterns", []) or [])[:3]],
        "failure_patterns": [_compact_item(item) for item in (review.get("failure_patterns", []) or [])[:3]],
        "governance_notes": [note for note in (review.get("governance_notes", []) or [])[:3] if note],
        "action_items": (review.get("action_items", []) or [])[:4],
        "memory_notes": (review.get("memory_notes", []) or [])[:3],
        "skill": review.get("skill", {}),
    }


def _goal_report_markdown(title: str, payload: dict[str, Any]) -> str:
    thresholds = payload.get("thresholds", {}) if isinstance(payload.get("thresholds", {}), dict) else {}
    window = payload.get("window", {}) if isinstance(payload.get("window", {}), dict) else {}
    gates = payload.get("gates", {}) if isinstance(payload.get("gates", {}), dict) else {}
    dd_gate = gates.get("max_drawdown_pct", {}) if isinstance(gates.get("max_drawdown_pct", {}), dict) else {}
    autonomy_gate = gates.get("autonomy_rate_pct", {}) if isinstance(gates.get("autonomy_rate_pct", {}), dict) else {}
    status = "PASS" if bool(gates.get("all_passed", False)) else "WATCH"
    return (
        f"# {title}\n\n"
        f"- Generated At: {payload.get('generated_at', '')}\n"
        f"- Status: {status}\n"
        f"- Thresholds: max_drawdown_pct<={thresholds.get('max_drawdown_pct', 10.0)}, autonomy_rate_pct>={thresholds.get('autonomy_rate_pct', 85.0)}\n\n"
        "## Core Metrics\n"
        f"- Realized PnL: {window.get('realized_pnl', 0.0)}\n"
        f"- Closed Trades: {window.get('closed_trades', window.get('closed_trade_count', 0))}\n"
        f"- Win Rate: {window.get('win_rate_pct', 0.0)}%\n"
        f"- Profit Factor: {window.get('profit_factor', 0.0)}\n"
        f"- Max Drawdown: {window.get('max_drawdown_pct', 0.0)}%\n"
        f"- Autonomy Rate: {window.get('autonomy_rate_pct', 'N/A')}%\n\n"
        "## Gate Checks\n"
        f"- Max Drawdown Gate: {dd_gate.get('current', 0.0)} <= {dd_gate.get('threshold', 10.0)} ({'PASS' if dd_gate.get('passed', False) else 'FAIL'})\n"
        f"- Autonomy Gate: {autonomy_gate.get('current', 0.0)} >= {autonomy_gate.get('threshold', 85.0)} ({'PASS' if autonomy_gate.get('passed', False) else 'FAIL'})\n"
    )


def refresh_goal_progress_reports() -> dict[str, Any]:
    try:
        from src.metrics_agg import aggregate_goal_targets
    except Exception:
        from metrics_agg import aggregate_goal_targets  # type: ignore

    target = aggregate_goal_targets(APP_DIR, "default")
    now = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
    weekly_payload = {
        "generated_at": now,
        "period": "weekly",
        "thresholds": target.get("thresholds", {}),
        "gates": target.get("gates", {}),
        "window": target.get("window_week", {}),
    }
    monthly_payload = {
        "generated_at": now,
        "period": "monthly",
        "thresholds": target.get("thresholds", {}),
        "gates": target.get("gates", {}),
        "window": target.get("window_month", {}),
    }
    _write_json(GOAL_PROGRESS_WEEKLY_JSON_FILE, weekly_payload)
    _write_json(GOAL_PROGRESS_MONTHLY_JSON_FILE, monthly_payload)
    _write_text(GOAL_PROGRESS_WEEKLY_MD_FILE, _goal_report_markdown("Goal Progress Weekly", weekly_payload))
    _write_text(GOAL_PROGRESS_MONTHLY_MD_FILE, _goal_report_markdown("Goal Progress Monthly", monthly_payload))
    return {"weekly": weekly_payload, "monthly": monthly_payload}


def refresh_weekly_review(force: bool = False) -> dict[str, Any]:
    if not force and _is_fresh(HERMES_WEEKLY_REVIEW_JSON_FILE, 3600):
        return _read_json(HERMES_WEEKLY_REVIEW_JSON_FILE)

    try:
        import gui_backend

        refresh_governance_brief("approval_chain", gui_backend.get_training_approval_chain())
        refresh_governance_brief("governance_lifecycle", gui_backend.get_governance_lifecycle())
    except Exception:
        pass

    review = _local_weekly_review()
    goal_reports = refresh_goal_progress_reports()
    monthly_goal = goal_reports.get("monthly", {}) if isinstance(goal_reports.get("monthly", {}), dict) else {}
    monthly_window = monthly_goal.get("window", {}) if isinstance(monthly_goal.get("window", {}), dict) else {}
    monthly_gates = monthly_goal.get("gates", {}) if isinstance(monthly_goal.get("gates", {}), dict) else {}
    review["goal_progress"] = monthly_goal
    review.setdefault("governance_notes", [])
    if isinstance(review.get("governance_notes"), list):
        review["governance_notes"].append(
            "终态门禁(月): "
            f"回撤 {monthly_window.get('max_drawdown_pct', 0.0)}% / 自治率 {monthly_window.get('autonomy_rate_pct', 0.0)}% "
            f"=> {'PASS' if monthly_gates.get('all_passed', False) else 'WATCH'}"
        )
    _compact_weekly_review_payload(review)
    prompt = f"""
只做三件事：
1. 用 `skill_manage` 读取或确认 `gold-weekend-review`。
2. 用 `trading_factors_query` 获取 `action=summary` 的交易摘要。
3. 用 `session_search` 搜索“交易失败 治理阻断 incomplete”最近相关结果。

只输出 JSON：
{{
  "generated_at": "...",
  "source": "hermes",
  "title": "Hermes Weekly Review",
  "executive_summary": [],
  "best_patterns": [],
  "failure_patterns": [],
  "governance_notes": [],
  "action_items": [],
  "memory_notes": [],
  "skill": {{
    "name": "{HERMES_SKILL_NAME}",
    "status": "created|existing|patched",
    "path": "{str(HERMES_SKILL_PATH)}",
    "summary": []
  }},
  "tool_context": {{
    "trading_summary": {{}},
    "session_hits": []
  }}
}}
""".strip()
    try:
        hermes_payload = _run_hermes_agent("gold-hermes-weekly-review-v2", prompt, timeout=90)
        if isinstance(hermes_payload, dict):
            for key in ("best_patterns", "failure_patterns", "memory_notes", "skill", "tool_context"):
                value = hermes_payload.get(key)
                if value not in (None, ""):
                    review[key] = value
            review["source"] = "local+hermes"
    except Exception as exc:
        review["hermes_error"] = str(exc)

    review["skill"] = _ensure_weekly_skill(review)
    _write_json(
        HERMES_SKILL_REGISTRY_FILE,
        {
            "generated_at": review.get("generated_at"),
            "source": review.get("source"),
            "skills": [review.get("skill", {})],
        },
    )
    _write_json(HERMES_WEEKLY_REVIEW_JSON_FILE, review)
    _write_text(HERMES_WEEKLY_REVIEW_FILE, _render_weekly_review_markdown(review))
    return review


def _main(argv: list[str]) -> int:
    if len(argv) < 2:
        return 0
    action = argv[1]
    if action == "trade":
        refresh_trade_post_event_snapshots(force=True)
        return 0
    if action == "governance" and len(argv) >= 4:
        kind = argv[2]
        payload_path = pathlib.Path(argv[3])
        payload = _read_json(payload_path)
        refresh_governance_brief(kind, payload)
        return 0
    if action == "weekly_review":
        refresh_weekly_review(force=True)
        return 0
    if action == "goal_reports":
        refresh_goal_progress_reports()
        return 0
    if action == "autonomy_cycle":
        try:
            from src.autonomy_loop import run_autonomy_loop_cycle
        except Exception:
            from autonomy_loop import run_autonomy_loop_cycle  # type: ignore
        run_autonomy_loop_cycle(APP_DIR, "default", "hermes_bridge")
        return 0
    return 0


if __name__ == "__main__":
    raise SystemExit(_main(sys.argv))
