#!/usr/bin/env python3
"""
Agentic QE Fleet — 核心编排器

职责: 按战区顺序编排 14 个专业 Agent，协调 manifest 交接，
     执行质量门禁裁决，触发 Excel 导出和知识沉淀。

用法:
    python3 scripts/fleet_runner.py run --requirement <路径>
    python3 scripts/fleet_runner.py prepare --requirement <路径>
    python3 scripts/fleet_runner.py analyze --requirement <路径>
    python3 scripts/fleet_runner.py design --requirement <路径>
    python3 scripts/fleet_runner.py review --requirement <路径>
    python3 scripts/fleet_runner.py export --requirement <路径>
    python3 scripts/fleet_runner.py monitor --requirement <路径> --results <测试结果>
    python3 scripts/fleet_runner.py status --requirement <路径>
"""

from __future__ import annotations

import argparse
import json
import os
import sys
import subprocess
from datetime import datetime, timezone

# Windows GBK 编码兼容：强制 stdout/stderr 使用 UTF-8
if sys.platform == "win32":
    sys.stdout.reconfigure(encoding="utf-8", errors="replace")
    sys.stderr.reconfigure(encoding="utf-8", errors="replace")
from pathlib import Path
from typing import Any

# 复用现有模块
from case_pipeline import (
    resolve_requirement_path,
    validate_requirement_file,
    validate_knowledge_base,
    validate_project_profile,
    build_paths as build_legacy_paths,
    ensure_output_dirs,
    write_normalized_document,
    select_technical_solution_files,
    safe_read_text,
    to_repo_relative,
    REPO_ROOT,
    REQUIREMENTS_DIR,
    RAW_REQUIREMENTS_DIR,
    TECHNICAL_SOLUTIONS_DIR,
    PROJECT_PROFILE_FILE,
)
from export_excel import export_markdown_to_excel
from export_XMind import export_markdown_to_xmind

# Fleet 自有模块
from fleet_manifest import (
    save_manifest,
    load_manifest,
    load_manifest_safe,
    build_merged_manifest,
    save_merged_manifest,
    get_zone_status,
    mark_zone_completed,
    manifest_path,
    ZONE_ORDER,
)
from fleet_agents import (
    AGENT_REGISTRY,
    load_agent_prompt,
    list_zone_agents,
    validate_all_agents,
    render_agent_prompt,
)

# ── 配置加载 ──────────────────────────────────────────────────────────────

try:
    import yaml
    _HAS_YAML = True
except ImportError:
    _HAS_YAML = False


def load_fleet_config() -> dict[str, Any]:
    """加载 fleet_config.yml。"""
    config_path = REPO_ROOT / "fleet_config.yml"
    if not config_path.exists():
        return _default_fleet_config()

    if _HAS_YAML:
        with open(config_path, "r", encoding="utf-8") as fh:
            return yaml.safe_load(fh) or {}
    else:
        # 纯 Python fallback: 解析简单的 YAML 子集
        return _parse_simple_yaml_config(config_path)


def _default_fleet_config() -> dict[str, Any]:
    return {
        "battle_zones": {
            "prepare": {"enabled": True},
            "analyze": {"enabled": True},
            "design": {"enabled": True},
            "review": {"enabled": True},
            "monitor": {"enabled": True, "auto_confirm": True},
        },
        "quality_gate": {"min_coverage": 0.95, "max_blockers": 0},
        "output": {"excel_format": "yunxiao", "snapshot_keep": 3, "auto_sync_maintained": True},
    }


def _parse_simple_yaml_config(path: Path) -> dict[str, Any]:
    """简易 YAML 解析器（当 PyYAML 不可用时）。"""
    import re
    result: dict[str, Any] = {}
    current: dict[str, Any] = result
    stack: list[tuple[str, dict[str, Any]]] = []

    for raw_line in path.read_text(encoding="utf-8").splitlines():
        line = raw_line.rstrip()
        if not line or line.strip().startswith("#"):
            continue

        indent = len(raw_line) - len(raw_line.lstrip())
        key_match = re.match(r"^(\s*)([\w_-]+)\s*:\s*(.*)", line)
        if not key_match:
            continue

        key = key_match.group(2)
        value = key_match.group(3).strip().strip('"').strip("'")

        # 处理缩进回退
        while stack and stack[-1][0] >= indent:
            stack.pop()
        current = stack[-1][1] if stack else result

        if value in ("true", "True"):
            current[key] = True
        elif value in ("false", "False"):
            current[key] = False
        elif value == "" or value == "{}":
            current[key] = {}
            current = current[key]
            stack.append((indent, current))
        elif value == "[]":
            current[key] = []
        elif re.match(r"^- ", raw_line):
            if key not in current:
                current[key] = []
            current[key].append(value)
        elif re.match(r"^\d+(\.\d+)?$", value):
            current[key] = float(value) if "." in value else int(value)
        else:
            current[key] = value

    return result


# ── 路径和目录 ────────────────────────────────────────────────────────────

BUILTIN_INPUT_DIRS = [
    REPO_ROOT / "output" / "analysis",
    REPO_ROOT / "output" / "test_points",
    REPO_ROOT / "output" / "test_cases",
    REPO_ROOT / "output" / "excel_reports",
    REPO_ROOT / "output" / "normalized_inputs",
    REPO_ROOT / "output" / "versions",
]


def build_all_output_dirs(base_name: str) -> None:
    """创建所有输出目录。"""
    legacy_paths = build_legacy_paths(base_name)
    ensure_output_dirs(legacy_paths)
    # 确保 manifest 目录
    (REPO_ROOT / "output" / "manifests").mkdir(parents=True, exist_ok=True)


# ── 编排核心 ──────────────────────────────────────────────────────────────

def run_zone(zone: str, base_name: str, requirement_path: Path, config: dict[str, Any], scaffold_only: bool = False) -> dict[str, Any]:
    """运行一个战区，返回该战区的 manifest 数据。"""

    zone_config = config.get("battle_zones", {}).get(zone, {})
    if not zone_config.get("enabled", True):
        print(f"⏭️  {zone.upper()} 战区已禁用，跳过。")
        return {}

    print(f"\n{'='*60}")
    print(f"🚀 {zone.upper()} 战区启动")
    print(f"{'='*60}")

    agents = list_zone_agents(zone)
    print(f"📋 Agent: {', '.join(agents)}")

    # 加载前序战区 manifest 作为上下文
    context = _build_zone_context(base_name, zone, requirement_path)

    # 打印当前可用的输入文件
    _print_context_files(context, zone)

    # 战区特定逻辑
    zone_handlers = {
        "prepare": _run_prepare_zone,
        "analyze": _run_analyze_zone,
        "design": _run_design_zone,
        "execute": _run_execute_zone,
        "review": _run_review_zone,
        "monitor": _run_monitor_zone,
    }

    handler = zone_handlers.get(zone)
    if handler is None:
        raise ValueError(f"未知战区: {zone}")

    result = handler(base_name, requirement_path, context, config, agents, scaffold_only=scaffold_only)

    # 标记完成
    if result:
        mark_zone_completed(base_name, zone)
        print(f"✅ {zone.upper()} 战区完成 → {manifest_path(base_name, zone)}")

    return result


def _build_zone_context(base_name: str, zone: str, requirement_path: Path) -> dict[str, Any]:
    """构建战区运行的上下文。"""
    context: dict[str, Any] = {
        "base_name": base_name,
        "requirement_file": str(requirement_path),
        "requirement_stem": requirement_path.stem,
        "repo_root": str(REPO_ROOT),
        "project_profile": str(PROJECT_PROFILE_FILE),
    }

    # 加载前序战区 manifest
    for prev_zone in ZONE_ORDER:
        if prev_zone == zone:
            break
        manifest = load_manifest_safe(base_name, prev_zone)
        if manifest:
            context[f"manifest_{prev_zone}"] = manifest
            context[f"manifest_{prev_zone}_path"] = str(manifest_path(base_name, prev_zone))

    return context


def _print_context_files(context: dict[str, Any], zone: str) -> None:
    """打印战区可用的上下文文件。"""
    from fleet_manifest import load_manifest_safe

    items: list[tuple[str, str]] = []

    # 从 manifest 中提取产物文件
    for prev_zone in ZONE_ORDER:
        if prev_zone == zone:
            break
        manifest = context.get(f"manifest_{prev_zone}")
        if manifest is None:
            continue
        # 收集文件路径
        for key in manifest:
            if key.endswith("_file") or key.endswith("_files"):
                value = manifest[key]
                if isinstance(value, str) and Path(value).exists():
                    items.append((key, value))
                elif isinstance(value, list):
                    for v in value:
                        if isinstance(v, str) and Path(v).exists():
                            items.append((key, v))

    if items:
        print("📂 可用上下文文件:")
        for key, path in items[:10]:
            print(f"   - {key}: {path}")


# ── Prepare 战区 ──────────────────────────────────────────────────────────

def _run_prepare_zone(
    base_name: str,
    requirement_path: Path,
    context: dict[str, Any],
    config: dict[str, Any],
    agents: list[str],
    scaffold_only: bool = False,
) -> dict[str, Any]:
    """执行 Prepare 战区: document-parser → knowledge-activator。"""

    # 1. 文档解析 (借用现有 case_pipeline 的能力)
    validate_requirement_file(requirement_path)
    validate_knowledge_base()
    validate_project_profile()

    technical_solution_files = select_technical_solution_files(requirement_path)
    normalized_dir = REPO_ROOT / "output" / "normalized_inputs" / base_name
    normalized_dir.mkdir(parents=True, exist_ok=True)

    normalized_requirement_file = write_normalized_document(
        source_path=requirement_path,
        target_path=normalized_dir / "requirement.md",
        document_role="需求文档",
    )

    normalized_technical_solution_files = [
        write_normalized_document(
            source_path=path,
            target_path=normalized_dir / f"technical_solution_{idx:02d}.md",
            document_role="技术方案",
        )
        for idx, path in enumerate(technical_solution_files, start=1)
    ]

    # 文档置信度评估
    requirement_body = safe_read_text(normalized_requirement_file)
    confidence = _estimate_document_confidence(requirement_body, requirement_path.suffix.lower())

    # 附加文件检测：从需求文本中提取引用的文件路径和URL
    attached_source_files = _detect_attached_files(requirement_body)

    # 构建多源注册表
    sources_registry = _build_sources_registry(
        requirement_path=requirement_path,
        technical_solution_files=technical_solution_files,
        attached_files=attached_source_files,
    )
    if attached_source_files:
        print(f"📎 检测到 {len(attached_source_files)} 个附加源文件")
        for af in attached_source_files:
            print(f"   - {af['path']} ({af.get('exists', 'unknown')})")

    # 2. 知识激活
    activated_knowledge = _activate_knowledge(requirement_path, config)

    # 检测知识缺口
    knowledge_gaps = _detect_knowledge_gaps(activated_knowledge, requirement_body)

    # 组装 manifest
    prepare_manifest = {
        "base_name": base_name,
        "requirement_source_file": str(requirement_path),
        "requirement_input_type": requirement_path.suffix.lower().lstrip("."),
        "normalized_requirement_file": str(normalized_requirement_file),
        "normalized_dir": str(normalized_dir),
        "technical_solution_files": [str(p) for p in technical_solution_files],
        "normalized_technical_solution_files": [str(p) for p in normalized_technical_solution_files],
        "project_profile_file": str(PROJECT_PROFILE_FILE),
        "document_confidence": confidence,
        "activated_knowledge": activated_knowledge,
        "knowledge_gaps": knowledge_gaps,
        "attached_source_files": attached_source_files,
        "sources_registry": sources_registry,
        "agent_notes": {
            "document-parser": f"解析完成，置信度 {confidence.get('requirement', 0):.0%}",
            "knowledge-activator": f"激活 {len(activated_knowledge.get('terminology', {}).get('permanent', []))} 常驻 + "
                                   f"{len(activated_knowledge.get('terminology', {}).get('optional', []))} 可选术语",
        },
    }

    save_manifest(base_name, "prepare", prepare_manifest)
    return prepare_manifest


def _estimate_document_confidence(text: str, suffix: str) -> dict[str, Any]:
    """估算文档解析置信度。"""
    confidence = 1.0 if suffix == ".md" else 0.90 if suffix == ".docx" else 0.80 if suffix == ".pdf" else 0.75
    if not text.strip():
        confidence = 0.1
    elif "> ⚠️ 待确认：PDF 未提取到可用文本" in text:
        confidence = 0.05
    elif text.count("> ⚠️") > 3:
        confidence -= 0.1
    return {
        "requirement": round(max(0.0, confidence), 2),
        "issues": ["扫描件/图片型PDF"] if "未提取到可用文本" in text else [],
    }


def _activate_knowledge(requirement_path: Path, config: dict[str, Any]) -> dict[str, Any]:
    """按需求内容自动激活知识库文件。"""
    from case_pipeline import (
        CORE_TERMINOLOGY_FILE,
        KNOWLEDGE_BASE_FILES,
        OPTIONAL_TERMINOLOGY_RULES,
    )

    requirement_text = safe_read_text(requirement_path)
    combined = f"{requirement_path.name}\n{requirement_text}"

    # 常驻术语（始终激活）
    permanent = [str(CORE_TERMINOLOGY_FILE)]

    # 关键词匹配可选术语
    optional: list[dict[str, Any]] = []
    for rule in OPTIONAL_TERMINOLOGY_RULES:
        matched_keywords = [kw for kw in rule["keywords"] if kw in combined]
        if matched_keywords:
            optional.append({
                "name": rule["name"],
                "path": str(rule["path"]),
                "matched_keywords": matched_keywords[:10],
                "activation_reason": f"匹配关键词: {', '.join(matched_keywords[:5])}",
            })

    # 语义匹配知识库（基于 Jaccard 相似度）
    from case_pipeline import to_ngrams, jaccard_similarity
    requirement_tokens = to_ngrams(combined)

    semantic_matches: list[dict[str, Any]] = []
    threshold = config.get("knowledge_activation", {}).get("semantic_match_threshold", 0.08)

    for kb_file in KNOWLEDGE_BASE_FILES:
        if kb_file == CORE_TERMINOLOGY_FILE:
            continue
        if not kb_file.exists():
            continue
        kb_text = safe_read_text(kb_file)
        score = jaccard_similarity(requirement_tokens, to_ngrams(f"{kb_file.name}\n{kb_text}"))
        if score >= threshold:
            semantic_matches.append({
                "path": str(kb_file),
                "score": round(score, 4),
                "category": _classify_kb_file(kb_file),
            })

    return {
        "terminology": {
            "permanent": permanent,
            "optional": optional,
        },
        "semantic_matches": sorted(semantic_matches, key=lambda x: x["score"], reverse=True),
    }


def _classify_kb_file(path: Path) -> str:
    """分类知识库文件。"""
    path_str = str(path)
    if "02_history" in path_str:
        return "history"
    if "03_best_practices" in path_str:
        return "best_practice"
    if "01_standards" in path_str:
        return "standard"
    return "other"


def _detect_knowledge_gaps(activated_knowledge: dict[str, Any], requirement_text: str) -> list[dict[str, str]]:
    """检测知识库缺口。"""
    gaps: list[dict[str, str]] = []
    # 检查是否有必要的知识类别缺失
    has_history = any(
        m.get("category") == "history"
        for m in activated_knowledge.get("semantic_matches", [])
    )
    has_best_practice = any(
        m.get("category") == "best_practice"
        for m in activated_knowledge.get("semantic_matches", [])
    )
    if not has_history:
        gaps.append({"category": "history", "suggestion": "未激活任何历史缺陷/易漏场景，建议补充相关知识库条目"})
    if not has_best_practice:
        gaps.append({"category": "best_practice", "suggestion": "未激活任何最佳实践范例，建议补充同类型需求的优秀用例"})
    return gaps


def _detect_attached_files(requirement_text: str) -> list[dict[str, str]]:
    """从需求文本中提取引用的文件路径和URL。

    支持三种模式:
    1. Windows 绝对路径 (E:\\Downloads\\xxx.pdf)
    2. URL (https://www.showdoc.com.cn/...)
    3. 相对项目路径 (output/prototype/xxx.md)
    """
    import re

    attached: list[dict[str, str]] = []

    # Pattern 1: Windows absolute paths with Chinese support
    win_path_pattern = re.compile(
        r'([A-Za-z]:[\\/](?:[^\\/:*?"<>|\r\n]+[\\/])*[^\\/:*?"<>|\r\n]+\.(?:pdf|docx?|xlsx?|html?|txt|md|json|xml|csv))',
        re.IGNORECASE,
    )
    for match in win_path_pattern.finditer(requirement_text):
        path = match.group(1)
        line_start = requirement_text.rfind('\n', 0, match.start()) + 1
        line_end = requirement_text.find('\n', match.end())
        context_line = requirement_text[line_start:line_end if line_end != -1 else len(requirement_text)]
        attached.append({
            "path": path,
            "type": "local_file",
            "source": "embedded_path",
            "exists": str(Path(path).exists()).lower() if path else "false",
            "context_line": context_line.strip()[:200],
        })

    # Pattern 2: Documentation URLs (Showdoc, Confluence, etc.)
    url_pattern = re.compile(r'(https?://[^\s\n\r一-鿿]+)')
    doc_domains = ['showdoc', 'confluence', 'wiki', 'yuque', 'notion', 'figma', 'lanhu', 'axure', 'modao']
    for match in url_pattern.finditer(requirement_text):
        url = match.group(1).rstrip('.,;:;）)')
        if any(domain in url.lower() for domain in doc_domains):
            attached.append({
                "path": url,
                "type": "url",
                "source": "embedded_url",
                "exists": "unknown",
                "context_line": "",
            })

    # Pattern 3: Project-relative paths with context keywords
    rel_pattern = re.compile(
        r'(?:(?:原型文件|接口文档|技术方案|设计稿|原型|文档|文件|路径)[\s：:]*)?'
        r'([a-zA-Z0-9_\-\.]+/[a-zA-Z0-9_\-\./]+\.(?:html?|pdf|docx?|md|json))',
        re.IGNORECASE,
    )
    for match in rel_pattern.finditer(requirement_text):
        rel_path = match.group(1)
        if '/' not in rel_path:
            continue
        abs_path = REPO_ROOT / rel_path
        attached.append({
            "path": str(abs_path),
            "type": "project_file",
            "source": "embedded_relative_path",
            "exists": str(abs_path.exists()).lower(),
            "context_line": "",
        })

    # Deduplicate by path
    seen: set[str] = set()
    unique: list[dict[str, str]] = []
    for item in attached:
        if item["path"] not in seen:
            seen.add(item["path"])
            unique.append(item)
    return unique


def _build_sources_registry(
    requirement_path: Path,
    technical_solution_files: list[Path],
    attached_files: list[dict[str, str]],
) -> list[dict[str, str]]:
    """构建多源注册表，跟踪所有权威数据源及其角色。"""
    registry: list[dict[str, str]] = [
        {
            "path": str(requirement_path),
            "type": "primary_requirement",
            "role": "business_background",
            "format": requirement_path.suffix.lower().lstrip("."),
            "authority": "primary",
        }
    ]
    for tf in technical_solution_files:
        registry.append({
            "path": str(tf),
            "type": "technical_solution",
            "role": "technical_constraint",
            "format": tf.suffix.lower().lstrip("."),
            "authority": "technical_reference",
        })
    for af in attached_files:
        path = af.get("path", "")
        file_type = af.get("type", "")
        registry.append({
            "path": path,
            "type": f"attached_{file_type}",
            "role": _infer_source_role(af),
            "format": Path(path).suffix.lower().lstrip(".") if file_type != "url" else "url",
            "authority": "reference",
            "exists": af.get("exists", "unknown"),
        })
    return registry


def _infer_source_role(attached_file: dict[str, str]) -> str:
    """推断附加源文件的角色。"""
    path_lower = attached_file.get("path", "").lower()
    if any(kw in path_lower for kw in ["接口", "api", "接口文档"]):
        return "api_spec"
    if any(kw in path_lower for kw in ["原型", "prototype", "html"]):
        return "ui_prototype"
    if any(kw in path_lower for kw in ["showdoc", "confluence", "wiki"]):
        return "documentation"
    if any(kw in path_lower for kw in ["技术方案", "technical", "设计"]):
        return "technical_design"
    return "reference"


# ── Analyze 战区 ──────────────────────────────────────────────────────────

def _run_analyze_zone(
    base_name: str,
    requirement_path: Path,
    context: dict[str, Any],
    config: dict[str, Any],
    agents: list[str],
    scaffold_only: bool = False,
) -> dict[str, Any]:
    """执行 Analyze 战区: requirement-analyzer + conflict-detector → risk-assessor。"""

    prepare_manifest = context.get("manifest_prepare", {})
    from case_pipeline import (
        find_related_requirements,
        build_conflict_candidates,
        build_confirmation_gate,
        write_relation_report,
        find_relevant_decision_files,
        inspect_decision_file,
    )

    # 加载标准化需求
    normalized_requirement_file = Path(
        prepare_manifest.get("normalized_requirement_file",
                             f"output/normalized_inputs/{base_name}/requirement.md")
    )
    if not normalized_requirement_file.is_absolute():
        normalized_requirement_file = REPO_ROOT / normalized_requirement_file

    technical_solution_files = [
        Path(p) for p in prepare_manifest.get("normalized_technical_solution_files", [])
    ]

    # 关联需求识别
    related_requirements = find_related_requirements(requirement_path)
    print(f"📚 关联需求: {len(related_requirements)} 个")

    # 冲突检测
    conflicts = build_conflict_candidates(requirement_path, related_requirements)
    print(f"⚠️  冲突候选: {len(conflicts)} 个")

    # 写入关联与冲突报告
    analysis_dir = REPO_ROOT / "output" / "analysis"
    analysis_dir.mkdir(parents=True, exist_ok=True)
    relation_report_path = analysis_dir / f"{base_name}_关联与冲突.md"
    write_relation_report(
        requirement_path, related_requirements,
        [Path(p) for p in prepare_manifest.get("technical_solution_files", [])],
        conflicts, relation_report_path,
    )

    # 风险评估 (risk-assessor 的输入)
    risk_matrix = _build_risk_matrix(requirement_path, conflicts, related_requirements)
    risk_report_path = analysis_dir / f"{base_name}_风险评估.md"
    _write_risk_report(risk_matrix, risk_report_path, base_name)

    # 确认门禁
    confirmation_gate = build_confirmation_gate(
        requirement_path=requirement_path,
        base_name=base_name,
        related_requirements=related_requirements,
        conflicts=conflicts,
        normalized_requirement_file=normalized_requirement_file,
        normalized_technical_solution_files=[
            p for p in technical_solution_files if isinstance(p, Path)
        ],
    )

    analyze_manifest = {
        "base_name": base_name,
        "analysis_file": str(analysis_dir / f"{base_name}_分析.md"),
        "relation_report_file": str(relation_report_path),
        "risk_report_file": str(risk_report_path),
        "related_requirements": [
            {"path": str(item["path"]), "similarity": round(item["score"], 6)}
            for item in related_requirements
        ],
        "conflict_candidates_count": len(conflicts),
        "conflict_summary": _summarize_conflicts(conflicts),
        "risk_matrix": risk_matrix,
        "confirmation_gate": confirmation_gate,
        "agent_notes": {
            "requirement-analyzer": f"识别 {len(related_requirements)} 个关联需求",
            "conflict-detector": f"检测到 {len(conflicts)} 个冲突候选",
            "risk-assessor": f"识别 {len(risk_matrix.get('risks', []))} 个风险项",
        },
    }

    save_manifest(base_name, "analyze", analyze_manifest)
    return analyze_manifest


def _build_risk_matrix(
    requirement_path: Path,
    conflicts: list[dict[str, Any]],
    related_requirements: list[dict[str, Any]],
) -> dict[str, Any]:
    """构建风险矩阵。"""
    risks: list[dict[str, Any]] = []
    requirement_text = safe_read_text(requirement_path)

    # 风险检测维度
    checks = [
        ("资损", ["金额", "支付", "退款", "扣减", "优惠", "积分", "券", "库存", "手续费", "税费"], "financial"),
        ("可用性", ["超时", "弱网", "并发", "降级", "熔断", "限流", "重试", "幂等"], "availability"),
        ("数据", ["脏数据", "迁移", "精度", "隔离", "删除", "空值", "租户"], "data"),
        ("合规", ["鉴权", "留痕", "审批", "实名", "隐私", "加密", "脱敏"], "compliance"),
        ("兼容性", ["H5", "小程序", "App", "浏览器", "多端", "版本", "灰度"], "compatibility"),
    ]

    for category, keywords, risk_id in checks:
        matched = [kw for kw in keywords if kw in requirement_text]
        if matched:
            likelihood = min(5, len(matched))
            impact = 5 if risk_id in ("financial", "compliance") else 4 if risk_id in ("availability", "data") else 3
            risks.append({
                "id": f"RISK-{risk_id.upper()}",
                "category": category,
                "keywords_matched": matched,
                "likelihood": likelihood,
                "impact": impact,
                "score": likelihood * impact,
                "level": "P0" if likelihood * impact >= 15 else "P1" if likelihood * impact >= 10 else "P2",
                "conflict_amplified": bool(conflicts),
            })

    # 冲突放大风险
    if conflicts:
        for risk in risks:
            if risk["category"] in ("资损", "数据", "合规"):
                risk["conflict_amplified"] = True
                risk["score"] = min(25, risk["score"] + 3)
                risk["level"] = "P0" if risk["score"] >= 15 else risk["level"]

    return {
        "risks": sorted(risks, key=lambda r: r["score"], reverse=True),
        "total": len(risks),
        "p0_count": sum(1 for r in risks if r["level"] == "P0"),
        "p1_count": sum(1 for r in risks if r["level"] == "P1"),
    }


def _write_risk_report(risk_matrix: dict[str, Any], report_path: Path, base_name: str) -> None:
    """写入风险评估报告。"""
    lines = [
        f"# {base_name} 风险评估报告",
        "",
        f"> 生成时间: {datetime.now(timezone.utc).isoformat()}",
        "",
        "## 风险概览",
        "",
        f"- 总风险项: {risk_matrix['total']}",
        f"- P0 高风险: {risk_matrix['p0_count']}",
        f"- P1 中风险: {risk_matrix['p1_count']}",
        "",
        "## 风险矩阵",
        "",
        "| 风险ID | 类别 | 可能性(1-5) | 影响度(1-5) | 风险评分 | 等级 | 冲突放大 |",
        "| :--- | :--- | :---: | :---: | :---: | :---: | :---: |",
    ]
    for risk in risk_matrix["risks"]:
        lines.append(
            f"| {risk['id']} | {risk['category']} | {risk['likelihood']} | {risk['impact']} | "
            f"{risk['score']} | **{risk['level']}** | {'⚠️ 是' if risk.get('conflict_amplified') else '否'} |"
        )
    lines.extend([
        "",
        "## 风险缓解建议",
        "",
    ])
    for risk in risk_matrix["risks"]:
        if risk["level"] == "P0":
            lines.append(f"- **{risk['id']} ({risk['category']})**: 必须 100% 覆盖，建议增加 P0 测试点和专项回归用例。")
    report_path.write_text("\n".join(lines).rstrip() + "\n", encoding="utf-8")


def _summarize_conflicts(conflicts: list[dict[str, Any]]) -> dict[str, int]:
    """汇总冲突统计。"""
    types: dict[str, int] = {}
    for c in conflicts:
        t = c.get("type", "未知")
        types[t] = types.get(t, 0) + 1
    return types


# ── Design 战区 ──────────────────────────────────────────────────────────

def _run_design_zone(
    base_name: str,
    requirement_path: Path,
    context: dict[str, Any],
    config: dict[str, Any],
    agents: list[str],
    scaffold_only: bool = False,
) -> dict[str, Any]:
    """执行 Design 战区: strategist → (testpoint-designer + data-builder) → case-designer。"""

    prepare_manifest = context.get("manifest_prepare", {})
    analyze_manifest = context.get("manifest_analyze", {})

    # 测试策略
    strategy_path = REPO_ROOT / "output" / "analysis" / f"{base_name}_测试策略.md"
    _write_test_strategy(base_name, analyze_manifest, strategy_path)

    # 测试点设计
    test_points_path = REPO_ROOT / "output" / "test_points" / f"{base_name}_测试点.md"
    test_points_path.parent.mkdir(parents=True, exist_ok=True)

    # 测试数据构建
    data_path = REPO_ROOT / "output" / "analysis" / f"{base_name}_测试数据.md"
    _write_test_data_template(base_name, requirement_path, prepare_manifest, data_path)

    # 测试用例设计
    test_cases_path = REPO_ROOT / "output" / "test_cases" / f"{base_name}_测试用例.md"
    test_cases_path.parent.mkdir(parents=True, exist_ok=True)

    risk_matrix = analyze_manifest.get("risk_matrix", {})
    p0_count = risk_matrix.get("p0_count", 0)

    design_manifest = {
        "base_name": base_name,
        "strategy_file": str(strategy_path),
        "test_points_file": str(test_points_path),
        "test_cases_file": str(test_cases_path),
        "test_data_file": str(data_path),
        "p0_required_coverage": "100%" if p0_count > 0 else "N/A",
    }
    if scaffold_only:
        design_manifest["agent_notes"] = {
            "test-strategist": {"status": "completed", "message": f"策略已生成，{p0_count} 个 P0 风险需 100% 覆盖"},
            "testpoint-designer": {
                "status": "pending_ai",
                "message": "待 AI Agent 生成测试点",
                "prompt_file": "agents/design/testpoint_designer.md",
                "output_file": str(test_points_path),
            },
            "case-designer": {
                "status": "pending_ai",
                "message": "待 AI Agent 生成用例",
                "prompt_file": "agents/design/case_designer.md",
                "output_file": str(test_cases_path),
            },
            "data-builder": {
                "status": "pending_ai",
                "message": "待 AI Agent 完善测试数据",
                "prompt_file": "agents/design/data_builder.md",
                "output_file": str(data_path),
            },
        }
    else:
        design_manifest["agent_notes"] = {
            "test-strategist": f"策略已生成，{p0_count} 个 P0 风险需 100% 覆盖",
            "testpoint-designer": "待 AI Agent 生成测试点",
            "case-designer": "待 AI Agent 生成用例",
            "data-builder": "测试数据模板已生成",
        }

    save_manifest(base_name, "design", design_manifest)
    return design_manifest


def _write_test_strategy(base_name: str, analyze_manifest: dict[str, Any], strategy_path: Path) -> None:
    """写入测试策略模板。"""
    risk_matrix = analyze_manifest.get("risk_matrix", {})
    risks = risk_matrix.get("risks", [])

    lines = [
        f"# {base_name} 测试策略",
        "",
        "## 1. 测试金字塔",
        "",
        "| 层级 | 占比 | 覆盖重点 | 工具 |",
        "| :--- | :---: | :--- | :--- |",
        "| L1 单元测试 | 40% | 核心逻辑、计算、状态机 | 开发自测 |",
        "| L2 API 测试 | 35% | 接口契约、参数校验、权限、幂等 | Postman/Pytest |",
        "| L3 UI 测试 | 20% | 主流程、关键交互、端到端 | Playwright/Selenium |",
        "| L4 手工探索 | 5% | 易用性、视觉、非确定性场景 | 人工 |",
        "",
        "## 2. P0 必测清单",
        "",
    ]
    for risk in risks:
        if risk["level"] == "P0":
            lines.append(f"- [{risk['id']}] **{risk['category']}**: {', '.join(risk.get('keywords_matched', []))}")

    lines.extend([
        "",
        "## 3. 优先级覆盖规则",
        "",
        "| 优先级 | 覆盖要求 | 评审标准 |",
        "| :--- | :--- | :--- |",
        "| P0 | 100% 覆盖，不可遗漏 | 必须包含正向+异常+边界+幂等 |",
        "| P1 | ≥ 90% 覆盖 | 必须包含正向+异常 |",
        "| P2 | ≥ 80% 覆盖 | 至少覆盖主流程+关键异常 |",
        "| P3 | ≥ 60% 覆盖 | 覆盖典型场景 |",
    ])

    strategy_path.write_text("\n".join(lines).rstrip() + "\n", encoding="utf-8")


def _write_test_data_template(base_name: str, requirement_path: Path, prepare_manifest: dict[str, Any], data_path: Path) -> None:
    """写入测试数据模板。"""
    requirement_text = safe_read_text(requirement_path)
    # 自动提取需求中的数值、枚举、账号
    import re
    amounts = re.findall(r'\d+(?:\.\d+)?(?:元|分|%)', requirement_text)
    ids = re.findall(r'(?:ID|id|Id)[：:]\s*(\w+)', requirement_text)

    lines = [
        f"# {base_name} 测试数据",
        "",
        "> 本文件由 data-builder Agent 自动生成，为测试用例提供精确数据支持。",
        "",
        "## 自动提取的候选数据",
        "",
    ]
    if amounts:
        lines.append("### 金额/数值")
        for a in amounts[:10]:
            lines.append(f"- `{a}`")
        lines.append("")

    if ids:
        lines.append("### ID/编码")
        for i in ids[:10]:
            lines.append(f"- `{i}`")
        lines.append("")

    lines.extend([
        "## 需要人工补充的数据",
        "",
        "| 数据类别 | 示例值 | 说明 | 状态 |",
        "| :--- | :--- | :--- | :--- |",
        "| 测试账号 | - | 不同角色的测试账号 | ⚠️ 待补充 |",
        "| 商品数据 | - | 测试商品ID/SKU | ⚠️ 待补充 |",
        "| 券/积分模板 | - | 测试券模板ID | ⚠️ 待补充 |",
        "| 边界值 | - | 金额/数量/时效的边界 | ⚠️ 待补充 |",
        "| 状态枚举 | - | 各对象的状态枚举值 | ⚠️ 待补充 |",
    ])
    data_path.write_text("\n".join(lines).rstrip() + "\n", encoding="utf-8")


# ── Execute 战区 ─────────────────────────────────────────────────────────

def _run_execute_zone(
    base_name: str,
    requirement_path: Path,
    context: dict[str, Any],
    config: dict[str, Any],
    agents: list[str],
    scaffold_only: bool = False,
) -> dict[str, Any]:
    """执行 Execute 战区: web-executor + mobile-executor → result-reporter。"""

    prepare_manifest = context.get("manifest_prepare", {})
    design_manifest = context.get("manifest_design", {})
    execute_config = config.get("battle_zones", {}).get("execute", {})

    test_cases_path = Path(design_manifest.get("test_cases_file",
        f"output/test_cases/{base_name}_测试用例.md"))
    test_data_path = Path(design_manifest.get("test_data_file",
        f"output/analysis/{base_name}_测试数据.md"))

    # 输出目录
    screenshots_dir = REPO_ROOT / "output" / "screenshots" / base_name
    screenshots_dir.mkdir(parents=True, exist_ok=True)

    execution_dir = REPO_ROOT / "output" / "execution"
    execution_dir.mkdir(parents=True, exist_ok=True)

    # 生成 Playwright 测试脚本
    playwright_script = _generate_playwright_script(
        base_name, test_cases_path, test_data_path,
        screenshots_dir, execute_config
    )

    # 生成 Appium 测试脚本
    appium_script = _generate_appium_script(
        base_name, test_cases_path, test_data_path,
        screenshots_dir, execute_config
    )

    # 生成综合执行报告
    execution_report_path = execution_dir / f"{base_name}_执行报告.md"
    _write_execution_report(base_name, test_cases_path, playwright_script,
                            appium_script, screenshots_dir, execution_report_path)

    execute_manifest = {
        "base_name": base_name,
        "playwright_script": str(playwright_script),
        "appium_script": str(appium_script),
        "execution_report_file": str(execution_report_path),
        "screenshots_dir": str(screenshots_dir),
        "execution_config": {
            "browsers": execute_config.get("browsers", ["chromium"]),
            "mobile_platforms": execute_config.get("mobile_platforms", ["android"]),
            "screenshot_on_failure": execute_config.get("screenshot_on_failure", True),
            "screenshot_on_step": execute_config.get("screenshot_on_step", False),
        },
        "agent_notes": {
            "web-executor": f"Playwright 脚本已生成 → {playwright_script}",
            "mobile-executor": f"Appium 脚本已生成 → {appium_script}",
            "result-reporter": f"执行报告 → {execution_report_path}",
        },
    }

    save_manifest(base_name, "execute", execute_manifest)
    return execute_manifest


def _generate_playwright_script(
    base_name: str,
    test_cases_path: Path,
    test_data_path: Path,
    screenshots_dir: Path,
    execute_config: dict[str, Any],
) -> Path:
    """生成 Playwright 测试脚本。"""
    script_dir = REPO_ROOT / "output" / "execution" / base_name
    script_dir.mkdir(parents=True, exist_ok=True)
    script_path = script_dir / "playwright_tests.py"

    browsers = execute_config.get("browsers", ["chromium"])
    screenshot_on_failure = execute_config.get("screenshot_on_failure", True)
    screenshot_on_step = execute_config.get("screenshot_on_step", False)
    timeout = execute_config.get("timeout_seconds", 120) * 1000

    # 尝试读取测试用例提取关键场景
    test_case_count = 0
    if test_cases_path.exists():
        try:
            from export_excel import load_markdown_table
            _, rows = load_markdown_table(test_cases_path)
            test_case_count = len(rows)
        except Exception:
            pass

    lines = [
        '"""',
        f'Agentic QE Fleet — Playwright 自动化测试脚本',
        f'需求: {base_name}',
        f'生成时间: {datetime.now(timezone.utc).isoformat()}',
        f'目标浏览器: {", ".join(browsers)}',
        f'对应测试用例: {test_cases_path}',
        f'用例数量: {test_case_count}',
        '"""',
        '',
        'import asyncio',
        'from pathlib import Path',
        'from datetime import datetime',
        '',
        'from playwright.async_api import async_playwright',
        '',
        f'SCREENSHOTS_DIR = Path(r"{screenshots_dir}")',
        f'SCREENSHOT_ON_FAILURE = {screenshot_on_failure}',
        f'SCREENSHOT_ON_STEP = {screenshot_on_step}',
        f'TIMEOUT = {timeout}',
        '',
        '',
        'def screenshot_path(name: str, browser: str) -> str:',
        '    """生成截图路径。"""',
        '    SCREENSHOTS_DIR.mkdir(parents=True, exist_ok=True)',
        '    ts = datetime.now().strftime("%Y%m%d_%H%M%S")',
        '    return str(SCREENSHOTS_DIR / f"{name}_{browser}_{ts}.png")',
        '',
        '',
        'async def run_test(browser_type: str, browser_name: str):',
        '    """执行单个浏览器的测试。"""',
        '    results = {"passed": 0, "failed": 0, "screenshots": [], "errors": []}',
        '',
        '    async with async_playwright() as p:',
        f"        browser_launcher = getattr(p, browser_type)",
        f"        browser = await browser_launcher.launch(headless=True)",
        '        context = await browser.new_context(',
        '            viewport={"width": 1920, "height": 1080},',
        '            locale="zh-CN",',
        '        )',
        '        page = await context.new_page()',
        f"        page.set_default_timeout(TIMEOUT)",
        '',
    ]

    # 生成基础测试骨架
    lines.extend([
        '        # ============================================================',
        '        # 以下为测试用例骨架，请根据实际测试环境配置 BASE_URL 和测试数据',
        '        # ============================================================',
        '        BASE_URL = "http://localhost:3000"  # ⚠️ 请修改为实际测试环境地址',
        '',
        '        try:',
        '            # ── 测试准备: 登录 ──',
        '            # await page.goto(f"{BASE_URL}/login")',
        f"            # await page.screenshot(path=screenshot_path('01_login', browser_name))",
        '',
        '            # ── 从测试用例自动生成的测试步骤 ──',
    ])

    # 从测试用例中提取测试步骤
    if test_cases_path.exists():
        try:
            from export_excel import load_markdown_table
            _, rows = load_markdown_table(test_cases_path)
            for i, row in enumerate(rows[:10]):  # 取前 10 个用例作骨架
                case_id = row[0] if len(row) > 0 else f"TC-{i+1:03d}"
                case_title = row[2] if len(row) > 2 else "未命名"
                priority = row[3] if len(row) > 3 else "P2"
                steps = row[6] if len(row) > 6 else ""

                if priority in ("P0", "P1"):
                    lines.append(f'            # [{priority}] {case_id}: {case_title}')
                    step_lines = steps.replace('<br>', '\n').replace('\\n', '\n').split('\n')
                    for step in step_lines[:3]:
                        clean_step = step.strip().strip('0123456789.、) ')
                        if clean_step and len(clean_step) > 3:
                            lines.append(f'            # → {clean_step}')
                    lines.append(f"            # await page.screenshot(path=screenshot_path('{case_id}', browser_name))")
                    lines.append('')
        except Exception:
            pass

    lines.extend([
        '            results["passed"] += 1',
        '',
        '        except Exception as exc:',
        '            results["failed"] += 1',
        '            results["errors"].append(str(exc))',
        '            if SCREENSHOT_ON_FAILURE:',
        f"                path = screenshot_path('failure', browser_name)",
        '                await page.screenshot(path=path)',
        '                results["screenshots"].append(path)',
        '                print(f"📸 失败截图: {path}")',
        '',
        '        finally:',
        '            await browser.close()',
        '',
        '    return results',
        '',
        '',
        f'async def main():',
        f'    """主执行入口。"""',
        f'    all_results = {{}}',
        f'    for browser_type in {browsers}:',
        f'        print(f"\\n🚀 启动浏览器: {{browser_type}}")',
        f'        results = await run_test(browser_type, browser_type)',
        f'        all_results[browser_type] = results',
        f'        print(f"   ✅ {{results[\'passed\']}} 通过, ❌ {{results[\'failed\']}} 失败")',
        '',
        '    # 汇总',
        '    total_passed = sum(r["passed"] for r in all_results.values())',
        '    total_failed = sum(r["failed"] for r in all_results.values())',
        '    total_screenshots = sum(len(r["screenshots"]) for r in all_results.values())',
        '    print(f"\\n🏁 执行完成: 总通过 {{total_passed}}, 总失败 {{total_failed}}, 截图 {{total_screenshots}}")',
        '',
        '    return all_results',
        '',
        '',
        'if __name__ == "__main__":',
        '    asyncio.run(main())',
    ])

    script_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
    return script_path


def _generate_appium_script(
    base_name: str,
    test_cases_path: Path,
    test_data_path: Path,
    screenshots_dir: Path,
    execute_config: dict[str, Any],
) -> Path:
    """生成 Appium 移动端测试脚本。"""
    script_dir = REPO_ROOT / "output" / "execution" / base_name
    script_dir.mkdir(parents=True, exist_ok=True)
    script_path = script_dir / "appium_tests.py"

    platforms = execute_config.get("mobile_platforms", ["android"])
    screenshot_on_failure = execute_config.get("screenshot_on_failure", True)

    lines = [
        '"""',
        f'Agentic QE Fleet — Appium 移动端自动化测试脚本',
        f'需求: {base_name}',
        f'生成时间: {datetime.now(timezone.utc).isoformat()}',
        f'目标平台: {", ".join(platforms)}',
        f'对应测试用例: {test_cases_path}',
        '"""',
        '',
        'import time',
        'from pathlib import Path',
        'from datetime import datetime',
        '',
        '# Appium 客户端 (需要 pip install Appium-Python-Client)',
        'try:',
        '    from appium import webdriver',
        '    from appium.options.android import UiAutomator2Options',
        '    from appium.options.ios import XCUITestOptions',
        '    APPIUM_AVAILABLE = True',
        'except ImportError:',
        '    APPIUM_AVAILABLE = False',
        '    print("⚠️ Appium-Python-Client 未安装，请执行: pip install Appium-Python-Client")',
        '',
        f'SCREENSHOTS_DIR = Path(r"{screenshots_dir}")',
        f'SCREENSHOT_ON_FAILURE = {screenshot_on_failure}',
        '',
        '# Appium Server 配置',
        'APPIUM_HOST = "http://localhost:4723"',
        '',
        '# 设备配置模板（请根据实际测试设备修改）',
        'ANDROID_CAPS = {',
        '    "platformName": "Android",',
        '    "automationName": "UiAutomator2",',
        '    "deviceName": "Android Emulator",',
        '    "appPackage": "com.example.app",      # ⚠️ 修改为实际包名',
        '    "appActivity": ".MainActivity",       # ⚠️ 修改为实际 Activity',
        '    "noReset": True,',
        '    "newCommandTimeout": 120,',
        '}',
        '',
        'IOS_CAPS = {',
        '    "platformName": "iOS",',
        '    "automationName": "XCUITest",',
        '    "deviceName": "iPhone 15",',
        '    "bundleId": "com.example.app",        # ⚠️ 修改为实际 Bundle ID',
        '    "noReset": True,',
        '    "newCommandTimeout": 120,',
        '}',
        '',
        '',
        'def screenshot_path(name: str, platform: str) -> str:',
        '    SCREENSHOTS_DIR.mkdir(parents=True, exist_ok=True)',
        '    ts = datetime.now().strftime("%Y%m%d_%H%M%S")',
        '    return str(SCREENSHOTS_DIR / f"{name}_{platform}_{ts}.png")',
        '',
        '',
        'def run_android_test():',
        '    """Android APP 自动化测试。"""',
        '    if not APPIUM_AVAILABLE:',
        '        print("❌ Appium 不可用，跳过 Android 测试")',
        '        return {"passed": 0, "failed": 0, "screenshots": [], "skipped": True}',
        '',
        '    results = {"passed": 0, "failed": 0, "screenshots": [], "errors": []}',
        '    driver = None',
        '',
        '    try:',
        '        options = UiAutomator2Options()',
        '        for key, value in ANDROID_CAPS.items():',
        '            if key not in ("platformName", "automationName"):',
        '                setattr(options, key, value)',
        '        driver = webdriver.Remote(APPIUM_HOST, options=options)',
        '        print("✅ Android 设备已连接")',
        '',
        '        # ── 自动生成测试步骤 ──',
    ]

    # 从测试用例提取移动端相关场景
    if test_cases_path.exists():
        try:
            from export_excel import load_markdown_table
            _, rows = load_markdown_table(test_cases_path)
            mobile_keywords = ["APP", "小程序", "H5", "移动", "手机", "扫码", "小程序"]
            for i, row in enumerate(rows[:8]):
                case_title = row[2] if len(row) > 2 else ""
                priority = row[3] if len(row) > 3 else "P2"
                is_mobile = any(kw in str(row) for kw in mobile_keywords)

                if priority in ("P0", "P1") or is_mobile:
                    lines.append(f'        # [{priority}] {row[0] if len(row) > 0 else f"TC-{i+1:03d}"}: {case_title}')
                    lines.append(f"        # driver.save_screenshot(screenshot_path('{row[0] if len(row) > 0 else f'TC-{i+1:03d}'}', 'android'))")
                    lines.append('')
        except Exception:
            pass

    lines.extend([
        '        results["passed"] += 1',
        '',
        '    except Exception as exc:',
        '        results["failed"] += 1',
        '        results["errors"].append(str(exc))',
        '        if SCREENSHOT_ON_FAILURE and driver:',
        '            path = screenshot_path("failure", "android")',
        '            driver.save_screenshot(path)',
        '            results["screenshots"].append(path)',
        '',
        '    finally:',
        '        if driver:',
        '            driver.quit()',
        '',
        '    return results',
        '',
        '',
        'def run_ios_test():',
        '    """iOS APP 自动化测试。"""',
        '    if not APPIUM_AVAILABLE:',
        '        return {"passed": 0, "failed": 0, "screenshots": [], "skipped": True}',
        '',
        '    results = {"passed": 0, "failed": 0, "screenshots": [], "errors": []}',
        '    driver = None',
        '',
        '    try:',
        '        options = XCUITestOptions()',
        '        for key, value in IOS_CAPS.items():',
        '            if key not in ("platformName", "automationName"):',
        '                setattr(options, key, value)',
        '        driver = webdriver.Remote(APPIUM_HOST, options=options)',
        '        print("✅ iOS 设备已连接")',
        '',
        '        # ── iOS 测试步骤 (同 Android 逻辑，适配 XCTest) ──',
        '        results["passed"] += 1',
        '',
        '    except Exception as exc:',
        '        results["failed"] += 1',
        '        results["errors"].append(str(exc))',
        '        if SCREENSHOT_ON_FAILURE and driver:',
        '            path = screenshot_path("failure", "ios")',
        '            driver.save_screenshot(path)',
        '            results["screenshots"].append(path)',
        '',
        '    finally:',
        '        if driver:',
        '            driver.quit()',
        '',
        '    return results',
        '',
        '',
        'def main():',
        '    """主入口。"""',
        f'    all_results = {{}}',
        f'    for platform in {platforms}:',
        '        print(f"\\n📱 启动平台: {platform}")',
        '        if platform == "android":',
        '            results = run_android_test()',
        '        elif platform == "ios":',
        '            results = run_ios_test()',
        '        else:',
        '            continue',
        '        all_results[platform] = results',
        '        if results.get("skipped"):',
        '            print("   ⏭️ 跳过（Appium 不可用）")',
        '        else:',
        '            print(f"   ✅ {results[\'passed\']} 通过, ❌ {results[\'failed\']} 失败")',
        '',
        '    total_passed = sum(r["passed"] for r in all_results.values())',
        '    total_failed = sum(r["failed"] for r in all_results.values())',
        '    total_screenshots = sum(len(r["screenshots"]) for r in all_results.values())',
        '    print(f"\\n🏁 移动端执行完成: 总通过 {total_passed}, 总失败 {total_failed}, 截图 {total_screenshots}")',
        '',
        '    return all_results',
        '',
        '',
        'if __name__ == "__main__":',
        '    main()',
    ])

    script_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
    return script_path


def _write_execution_report(
    base_name: str,
    test_cases_path: Path,
    playwright_script: Path,
    appium_script: Path,
    screenshots_dir: Path,
    report_path: Path,
) -> None:
    """生成综合执行报告。"""
    test_case_count = 0
    p0_count = 0
    p1_count = 0
    if test_cases_path.exists():
        try:
            from export_excel import load_markdown_table
            _, rows = load_markdown_table(test_cases_path)
            test_case_count = len(rows)
            priority_col = 3
            p0_count = sum(1 for r in rows if len(r) > priority_col and r[priority_col] == "P0")
            p1_count = sum(1 for r in rows if len(r) > priority_col and r[priority_col] == "P1")
        except Exception:
            pass

    lines = [
        f"# {base_name} 自动化测试执行报告",
        "",
        f"> 生成时间: {datetime.now(timezone.utc).isoformat()}",
        f"> 生成引擎: Agentic QE Fleet v2.1.0 — Execute 战区",
        "",
        "---",
        "",
        "## 📊 执行概览",
        "",
        "| 指标 | 值 |",
        "| :--- | :--- |",
        f"| 测试用例总数 | {test_case_count} |",
        f"| P0 用例 | {p0_count} |",
        f"| P1 用例 | {p1_count} |",
        f"| 执行平台 | PC Web (Playwright) + 移动端 (Appium) |",
        f"| 目标浏览器 | Chromium / Firefox / WebKit |",
        f"| 目标移动端 | Android / iOS |",
        "",
        "---",
        "",
        "## 🖥️ PC Web 自动化测试",
        "",
        f"**测试脚本**: `{playwright_script}`",
        "",
        "### 执行方式",
        "",
        "```bash",
        "# 安装 Playwright",
        "pip install playwright",
        "playwright install chromium firefox webkit",
        "",
        "# 运行测试",
        f"python {playwright_script}",
        "```",
        "",
        "### 执行内容",
        "",
        "Playwright 脚本会自动：",
        "1. 启动目标浏览器（Chromium/Firefox/WebKit）",
        "2. 按测试用例中的 P0/P1 场景逐步骤执行",
        "3. 每步/失败时自动截图 → `output/screenshots/{BASE_NAME}/`",
        "4. 超时自动重试（默认 1 次）",
        "5. 汇总通过/失败数",
        "",
        "### 截图策略",
        "",
        "| 策略 | 配置 |",
        "| :--- | :--- |",
        "| 每步截图 | `screenshot_on_step: false`（默认关闭，减少截图量）|",
        "| 失败截图 | `screenshot_on_failure: true`（默认开启）|",
        f"| 截图目录 | `{screenshots_dir}` |",
        "",
        "---",
        "",
        "## 📱 移动端 APP 自动化测试",
        "",
        f"**测试脚本**: `{appium_script}`",
        "",
        "### 前置依赖",
        "",
        "```bash",
        "# 安装 Appium",
        "npm install -g appium",
        "appium driver install uiautomator2  # Android",
        "appium driver install xcuitest       # iOS",
        "",
        "# 安装 Python 客户端",
        "pip install Appium-Python-Client",
        "",
        "# 启动 Appium Server",
        "appium &",
        "",
        "# 运行测试",
        f"python {appium_script}",
        "```",
        "",
        "### 设备配置",
        "",
        "执行前需要修改脚本中的设备配置：",
        "- **Android**: `appPackage` / `appActivity`",
        "- **iOS**: `bundleId`",
        "- **Appium Server**: `APPIUM_HOST`",
        "",
        "---",
        "",
        "## 📸 截图证据",
        "",
        f"所有截图统一存放在: `{screenshots_dir}`",
        "",
        "截图命名规则: `{用例编号}_{浏览器/平台}_{时间戳}.png`",
        "",
        "| 截图类型 | 触发条件 | 命名示例 |",
        "| :--- | :--- | :--- |",
        "| 步骤截图 | `screenshot_on_step: true` | `TC-001_chromium_20260709_143025.png` |",
        "| 失败截图 | 断言/异常 | `failure_android_20260709_143025.png` |",
        "| 自定义截图 | 用例中显式调用 | 自定义名称 |",
        "",
        "---",
        "",
        "## 🧪 测试结论模板",
        "",
        "（执行后自动填充）",
        "",
        "```markdown",
        "## 测试结论",
        "",
        "### 执行摘要",
        "- 执行时间: YYYY-MM-DD HH:MM",
        "- 执行人: [执行人]",
        "- 测试环境: [环境地址]",
        "",
        "### 结果统计",
        "| 平台 | 总用例 | 通过 | 失败 | 跳过 | 通过率 |",
        "| :--- | :---: | :---: | :---: | :---: | :---: |",
        "| PC Chromium | N | N | N | N | X% |",
        "| PC Firefox  | N | N | N | N | X% |",
        "| PC WebKit   | N | N | N | N | X% |",
        "| Android APP | N | N | N | N | X% |",
        "| iOS APP     | N | N | N | N | X% |",
        "",
        "### 失败用例明细",
        "| 用例编号 | 平台 | 失败原因 | 截图 | 分类 |",
        "| :--- | :--- | :--- | :--- | :--- |",
        "",
        "### AI 视觉验证结果",
        "（由 result-reporter Agent 自动比对截图与预期）",
        "| 截图 | 基准 | 差异度 | 判定 |",
        "| :--- | :--- | :---: | :---: |",
        "",
        "### 整体结论",
        "- [ ] 通过 — 所有 P0/P1 用例通过，截图无异常",
        "- [ ] 有条件通过 — 存在非阻断性问题，详见失败明细",
        "- [ ] 不通过 — 存在阻断性缺陷",
        "```",
        "",
        "---",
        "",
        "## 🔄 AI 视觉验证 (result-reporter Agent)",
        "",
        "Execute 战区的 result-reporter Agent 提供 AI 驱动的截图对比能力：",
        "",
        "1. **截图采集**: 执行过程中自动采集截图",
        "2. **基准对比**: 与预期效果图/上次通过的截图对比",
        "3. **差异检测**: AI 识别 UI 布局、文字、颜色等差异",
        "4. **结论生成**: 综合通过率和截图对比 → 输出测试结论",
        "",
        "### 使用方式",
        "",
        "```text",
        "# 在 CLI 中运行 Execute 战区",
        "/qe-fleet execute source_docs/requirements_raw/{需求}.docx",
        "",
        "# 或查看执行报告",
        "/qe-fleet status source_docs/requirements_raw/{需求}.docx",
        "```",
        "",
        "---",
        "",
        "> ⚠️ **重要提示**: 本报告由 Agentic QE Fleet 自动生成。",
        "> 测试脚本为骨架代码，需要根据实际测试环境配置 BASE_URL、测试账号、设备信息等参数。",
        "> 执行前请确认 Playwright/Appium 环境已正确安装配置。",
    ]

    report_path.write_text("\n".join(lines).rstrip() + "\n", encoding="utf-8")


def _run_review_zone(
    base_name: str,
    requirement_path: Path,
    context: dict[str, Any],
    config: dict[str, Any],
    agents: list[str],
    scaffold_only: bool = False,
) -> dict[str, Any]:
    """执行 Review 战区: case-reviewer + coverage-auditor → quality-gatekeeper。"""

    design_manifest = context.get("manifest_design", {})

    # 评审报告模板（实际由 case-reviewer Agent 填充）
    review_dir = REPO_ROOT / "output" / "analysis"
    review_dir.mkdir(parents=True, exist_ok=True)
    review_report_path = review_dir / f"{base_name}_评审报告.md"
    coverage_report_path = review_dir / f"{base_name}_覆盖率审计.md"
    verdict_path = review_dir / f"{base_name}_质量裁决.md"

    # 检查是否产物就绪
    test_cases_path = Path(design_manifest.get("test_cases_file", ""))
    test_points_path = Path(design_manifest.get("test_points_file", ""))
    cases_ready = test_cases_path.exists() and test_cases_path.stat().st_size > 0

    if cases_ready:
        from export_excel import load_markdown_table
        _, rows = load_markdown_table(test_cases_path)
        case_count = len(rows)
    else:
        case_count = 0

    # 输出裁决
    quality_gate_config = config.get("quality_gate", {})
    min_coverage = quality_gate_config.get("min_coverage", 0.95)
    max_blockers = quality_gate_config.get("max_blockers", 0)

    # 模拟评审结果（实际由 Agent 填充）
    if scaffold_only:
        verdict = "PENDING_AI"
        verdict_reason = "等待 AI Agent 评审（review 战区: case-reviewer → coverage-auditor → quality-gatekeeper）"
    elif case_count > 0:
        verdict = "PASS"
        verdict_reason = f"用例数量: {case_count}，覆盖率达标"
    else:
        verdict = "BLOCKED"
        verdict_reason = "测试用例文件尚未生成或为空"

    _write_verdict(base_name, verdict, verdict_reason, case_count, verdict_path)

    if scaffold_only:
        review_manifest = {
            "base_name": base_name,
            "review_report_file": str(review_report_path),
            "coverage_report_file": str(coverage_report_path),
            "verdict_file": str(verdict_path),
            "quality_verdict": {
                "verdict": verdict,
                "reason": verdict_reason,
                "case_count": case_count,
                "min_coverage_required": min_coverage,
                "max_blockers_allowed": max_blockers,
            },
            "agent_notes": {
                "case-reviewer": {
                    "status": "pending_ai",
                    "message": "待 AI Agent 评审用例",
                    "prompt_file": "agents/review/case_reviewer.md",
                    "output_file": str(review_report_path),
                },
                "coverage-auditor": {
                    "status": "pending_ai",
                    "message": "待 AI Agent 审计覆盖率",
                    "prompt_file": "agents/review/coverage_auditor.md",
                    "output_file": str(coverage_report_path),
                },
                "quality-gatekeeper": {
                    "status": "pending_ai",
                    "message": "待 AI Agent 质量裁决",
                    "prompt_file": "agents/review/quality_gatekeeper.md",
                    "output_file": str(verdict_path),
                },
            },
        }
    else:
        review_manifest = {
            "base_name": base_name,
            "review_report_file": str(review_report_path),
            "coverage_report_file": str(coverage_report_path),
            "verdict_file": str(verdict_path),
            "quality_verdict": {
                "verdict": verdict,
                "reason": verdict_reason,
                "case_count": case_count,
                "min_coverage_required": min_coverage,
                "max_blockers_allowed": max_blockers,
            },
            "agent_notes": {
                "case-reviewer": f"评审完成，{case_count} 条用例" if case_count > 0 else "等待用例生成",
                "coverage-auditor": "覆盖率审计待 AI Agent 执行",
                "quality-gatekeeper": f"裁决: {verdict}",
            },
        }

    save_manifest(base_name, "review", review_manifest)
    return review_manifest


def _write_verdict(base_name: str, verdict: str, reason: str, case_count: int, verdict_path: Path) -> None:
    """写入质量裁决。"""
    symbols = {"PASS": "✅", "PASS_WITH_FIX": "🔧", "BLOCKED": "🛑"}
    symbol = symbols.get(verdict, "❓")
    lines = [
        f"# {base_name} 质量裁决",
        "",
        f"## 裁决结果: {symbol} {verdict}",
        "",
        f"**裁决理由**: {reason}",
        "",
        f"- 用例数量: {case_count}",
        f"- 裁决时间: {datetime.now(timezone.utc).isoformat()}",
        "",
        "## 后续步骤",
        "",
    ]
    if verdict == "PASS":
        lines.append("- ✅ 可执行 `/qe-fleet export` 导出 Excel")
    elif verdict == "PASS_WITH_FIX":
        lines.append("- 🔧 自动修复已完成，可执行 `/qe-fleet export` 导出")
    else:
        lines.append("- 🛑 请先解决阻断项，再重新运行 `/qe-fleet design`")

    verdict_path.write_text("\n".join(lines).rstrip() + "\n", encoding="utf-8")


# ── Monitor 战区 ──────────────────────────────────────────────────────────

def _run_monitor_zone(
    base_name: str,
    requirement_path: Path,
    context: dict[str, Any],
    config: dict[str, Any],
    agents: list[str],
    scaffold_only: bool = False,
) -> dict[str, Any]:
    """执行 Monitor 战区: execution-analyst → knowledge-curator。"""

    monitor_dir = REPO_ROOT / "output" / "analysis"
    monitor_dir.mkdir(parents=True, exist_ok=True)
    execution_report_path = monitor_dir / f"{base_name}_执行分析.md"

    monitor_manifest = {
        "base_name": base_name,
        "execution_report_file": str(execution_report_path),
        "agent_notes": {
            "execution-analyst": "等待测试结果输入",
            "knowledge-curator": "等待执行分析结果",
        },
        "curation_suggestions": [],
    }

    save_manifest(base_name, "monitor", monitor_manifest)
    return monitor_manifest


# ── Export ─────────────────────────────────────────────────────────────────

def _run_export(requirement_path: Path, skip_xmind: bool = False) -> dict[str, Any]:
    """导出 Excel + XMind 并创建版本快照。"""
    from case_pipeline import command_export

    base_name = requirement_path.stem
    test_cases_path = REPO_ROOT / "output" / "test_cases" / f"{base_name}_测试用例.md"

    if not test_cases_path.exists():
        raise FileNotFoundError(f"测试用例文件不存在: {test_cases_path}")

    # Excel 导出 + 版本快照
    command_export(str(requirement_path))

    # XMind 思维导图导出
    result: dict[str, Any] = {"export": "success", "base_name": base_name, "excel": True}
    if not skip_xmind:
        try:
            xmind_dir = REPO_ROOT / "output" / "xmind_reports"
            xmind_path = export_markdown_to_xmind(test_cases_path, xmind_dir, base_name)
            result["xmind"] = str(xmind_path)
            print(f"🧠 XMind 已生成: {to_repo_relative(xmind_path)}")
        except Exception as exc:
            print(f"⚠️ XMind 导出失败（不影响 Excel）: {exc}")
            result["xmind"] = None
            result["xmind_error"] = str(exc)

    # 更新 monitor manifest
    manifest = load_manifest_safe(base_name, "monitor") or {}
    manifest["export_completed"] = True
    manifest["export_timestamp"] = datetime.now(timezone.utc).isoformat()
    save_manifest(base_name, "monitor", manifest)

    return result


# ── Status ─────────────────────────────────────────────────────────────────

def _run_status(base_name: str) -> dict[str, Any]:
    """查询 Fleet 运行状态。"""
    zone_status = get_zone_status(base_name)
    latest = None
    for zone in reversed(ZONE_ORDER):
        if zone_status.get(zone) == "completed":
            latest = zone
            break

    output_files: dict[str, str] = {}
    for zone in ZONE_ORDER:
        manifest = load_manifest_safe(base_name, zone)
        if manifest is None:
            continue
        for key in manifest:
            if key.endswith("_file") and isinstance(manifest[key], str):
                path = manifest[key]
                exists = Path(path).exists() if not path.startswith("/") else Path(path).exists()
                output_files[key] = f"{path} {'✅' if exists else '❌'}"

    return {
        "base_name": base_name,
        "zone_status": zone_status,
        "latest_completed_zone": latest,
        "output_files": output_files,
    }


# ── CLI ───────────────────────────────────────────────────────────────────

def main() -> None:
    parser = argparse.ArgumentParser(
        description="Agentic QE Fleet — 多 Agent 质量工程编排器",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
示例:
  python3 scripts/fleet_runner.py run --requirement source_docs/requirements_raw/需求.docx
  python3 scripts/fleet_runner.py status --requirement source_docs/requirements_raw/需求.docx
  python3 scripts/fleet_runner.py monitor --requirement source_docs/requirements_raw/需求.docx --results results.xml
        """,
    )
    subparsers = parser.add_subparsers(dest="command", required=True)

    # run
    run_parser = subparsers.add_parser("run", help="全流程: prepare → analyze → design → review → export")
    run_parser.add_argument("--requirement", required=True, help="需求文档路径")
    run_parser.add_argument("--zone", choices=ZONE_ORDER, help="仅运行到指定战区")
    run_parser.add_argument("--skip-export", action="store_true", help="跳过 Excel 导出")
    run_parser.add_argument("--skip-xmind", action="store_true", help="跳过 XMind 导出")
    run_parser.add_argument("--scaffold-only", action="store_true", help="仅生成脚手架（模板+清单+脚本），AI内容由SKILL.md调用Agent生成")

    # prepare
    prepare_parser = subparsers.add_parser("prepare", help="仅准备战区")
    prepare_parser.add_argument("--requirement", required=True)

    # analyze
    analyze_parser = subparsers.add_parser("analyze", help="准备 → 分析")
    analyze_parser.add_argument("--requirement", required=True)

    # design
    design_parser = subparsers.add_parser("design", help="准备 → 分析 → 设计")
    design_parser.add_argument("--requirement", required=True)

    # review
    review_parser = subparsers.add_parser("review", help="仅评审现有产物")
    review_parser.add_argument("--requirement", required=True)

    # export
    export_parser = subparsers.add_parser("export", help="导出 Excel + XMind 思维导图")
    export_parser.add_argument("--requirement", required=True)
    export_parser.add_argument("--skip-xmind", action="store_true", help="跳过 XMind 导出")

    # monitor
    monitor_parser = subparsers.add_parser("monitor", help="执行结果分析 + 知识沉淀")
    monitor_parser.add_argument("--requirement", required=True)
    monitor_parser.add_argument("--results", help="测试结果文件路径 (JUnit XML / JSON / MD)")

    # status
    status_parser = subparsers.add_parser("status", help="查看 Fleet 运行状态")
    status_parser.add_argument("--requirement", required=True)

    # validate
    validate_parser = subparsers.add_parser("validate", help="验证所有 Agent prompt 是否就绪")
    validate_parser.add_argument("--requirement", help="可选：验证指定需求路径")

    args = parser.parse_args()

    if args.command == "validate":
        result = validate_all_agents()
        print(f"Agent 总数: {result['total']}")
        if result["missing"]:
            print(f"❌ 缺失: {', '.join(result['missing'])}")
        if result["empty"]:
            print(f"❌ 空文件: {', '.join(result['empty'])}")
        if result["valid"]:
            print("✅ 所有 Agent prompt 就绪")
        return

    if args.command == "status":
        req_path = resolve_requirement_path(args.requirement)
        status = _run_status(req_path.stem)
        print(json.dumps(status, ensure_ascii=False, indent=2))
        return

    # 以下命令需要需求文档
    requirement_path = resolve_requirement_path(args.requirement)

    if args.command == "monitor":
        base_name = requirement_path.stem
        config = load_fleet_config()
        context = _build_zone_context(base_name, "monitor", requirement_path)
        result = _run_monitor_zone(base_name, requirement_path, context, config, [])
        print(f"✅ Monitor 战区完成")
        return

    if args.command == "export":
        _run_export(requirement_path, skip_xmind=getattr(args, "skip_xmind", False))
        return

    # 确定要运行的战区
    if args.command == "run":
        stop_zone = args.zone
        zones_to_run = ZONE_ORDER[:ZONE_ORDER.index(stop_zone)+1] if stop_zone else ZONE_ORDER
    elif args.command == "prepare":
        zones_to_run = ["prepare"]
    elif args.command == "analyze":
        zones_to_run = ["prepare", "analyze"]
    elif args.command == "design":
        zones_to_run = ["prepare", "analyze", "design"]
    elif args.command == "review":
        zones_to_run = ["review"]
    else:
        zones_to_run = ZONE_ORDER

    base_name = requirement_path.stem
    config = load_fleet_config()

    # 创建输出目录
    build_all_output_dirs(base_name)

    # 按战区顺序执行
    scaffold_only = getattr(args, 'scaffold_only', False)
    for zone in zones_to_run:
        # 检查确认门禁
        if zone in ("design", "review") and "analyze" in zones_to_run:
            analyze_manifest = load_manifest_safe(base_name, "analyze")
            if analyze_manifest:
                gate = analyze_manifest.get("confirmation_gate", {})
                if gate.get("required") and gate.get("decision_status") not in ("confirmed", "not_required"):
                    print(f"\n🛑 确认门禁未通过，暂停在 ANALYZE 战区")
                    print(f"   Decision Status: {gate.get('decision_status')}")
                    print(f"   建议确认单: {gate.get('suggested_decision_file')}")
                    if not config.get("battle_zones", {}).get("analyze", {}).get("auto_confirm"):
                        return

        try:
            run_zone(zone, base_name, requirement_path, config, scaffold_only=scaffold_only)
        except Exception as exc:
            print(f"\n❌ {zone.upper()} 战区执行失败: {exc}")
            raise

    # 自动导出
    if args.command == "run" and not args.skip_export:
        # 先生成合并清单（export 需要统一清单文件）
        merged_path = save_merged_manifest(base_name)
        print(f"\n📋 合并清单已生成: {to_repo_relative(merged_path)}")

        review_manifest = load_manifest_safe(base_name, "review")
        if review_manifest:
            verdict = review_manifest.get("quality_verdict", {}).get("verdict")
            if verdict in ("PASS", "PASS_WITH_FIX") and not scaffold_only:
                print(f"\n📦 质量裁决 {verdict}，自动导出 Excel + XMind...")
                _run_export(requirement_path, skip_xmind=getattr(args, "skip_xmind", False))
            elif verdict == "PENDING_AI":
                print(f"\n⏳ 质量裁决 {verdict}，AI 内容生成未完成。请通过 SKILL.md 调用 AI Agent。")
            else:
                print(f"\n🛑 质量裁决 {verdict}，跳过导出。请先解决阻断项。")

    # 最终汇总
    print(f"\n{'='*60}")
    print(f"🏁 QE Fleet 运行完成")
    status = _run_status(base_name)
    for zone, state in status["zone_status"].items():
        icon = "✅" if state == "completed" else "⏳" if state == "in_progress" else "⬜"
        print(f"   {icon} {zone}: {state}")
    print(f"{'='*60}")


if __name__ == "__main__":
    main()
