Yb-QaAutomationHub/scripts/fleet_agents.py

"""
Agentic QE Fleet — Agent 加载和上下文注入模块

负责:
1. 加载 Agent prompt 文件
2. 注入战区上下文（前序 manifest 数据）
3. 渲染最终可执行的 Agent prompt
"""

from __future__ import annotations

import re
from pathlib import Path
from typing import Any

REPO_ROOT = Path(__file__).resolve().parent.parent
AGENTS_DIR = REPO_ROOT / "agents"

# Agent 注册表: agent_id → (战区, 文件路径, 描述)
AGENT_REGISTRY: dict[str, dict[str, Any]] = {
    # ── Prepare ──
    "document-parser": {
        "zone": "prepare",
        "path": "prepare/document_parser.md",
        "name": "文档解析专家",
        "description": "解析原始需求文档为标准 Markdown，自动识别技术方案",
    },
    "knowledge-activator": {
        "zone": "prepare",
        "path": "prepare/knowledge_activator.md",
        "name": "知识激活专家",
        "description": "按需求内容自动激活术语/规则/历史缺陷/最佳实践",
    },
    # ── Analyze ──
    "requirement-analyzer": {
        "zone": "analyze",
        "path": "analyze/requirement_analyzer.md",
        "name": "需求分析专家",
        "description": "结构化需求模型 + 歧义标注",
    },
    "conflict-detector": {
        "zone": "analyze",
        "path": "analyze/conflict_detector.md",
        "name": "冲突检测专家",
        "description": "语义级历史需求规则冲突检测",
    },
    "risk-assessor": {
        "zone": "analyze",
        "path": "analyze/risk_assessor.md",
        "name": "风险评估专家",
        "description": "多维度风险矩阵量化",
    },
    # ── Design ──
    "test-strategist": {
        "zone": "design",
        "path": "design/test_strategist.md",
        "name": "测试策略师",
        "description": "分层测试策略 + 优先级矩阵",
    },
    "testpoint-designer": {
        "zone": "design",
        "path": "design/testpoint_designer.md",
        "name": "测试点设计师",
        "description": "全面测试点矩阵 + 来源标注",
    },
    "case-designer": {
        "zone": "design",
        "path": "design/case_designer.md",
        "name": "用例设计师",
        "description": "可执行测试用例 + 双验证预期结果",
    },
    "data-builder": {
        "zone": "design",
        "path": "design/data_builder.md",
        "name": "数据构造师",
        "description": "精确测试数据集构造",
    },
    # ── Review ──
    "case-reviewer": {
        "zone": "review",
        "path": "review/case_reviewer.md",
        "name": "用例评审师",
        "description": "用例质量/规范性/可执行性逐项评审",
    },
    "coverage-auditor": {
        "zone": "review",
        "path": "review/coverage_auditor.md",
        "name": "覆盖率审计师",
        "description": "需求→测试点→用例三级追溯覆盖审计",
    },
    "quality-gatekeeper": {
        "zone": "review",
        "path": "review/quality_gatekeeper.md",
        "name": "质量门禁裁决官",
        "description": "三级裁决: PASS / PASS_WITH_FIX / BLOCKED",
    },
    # ── Monitor ──
    "execution-analyst": {
        "zone": "monitor",
        "path": "monitor/execution_analyst.md",
        "name": "执行结果分析师",
        "description": "失败归类 + 模式识别 + 根因推测",
    },
    "knowledge-curator": {
        "zone": "monitor",
        "path": "monitor/knowledge_curator.md",
        "name": "知识沉淀师",
        "description": "自动回写知识库 + 去重保护",
    },
}


def get_agent_path(agent_id: str) -> Path:
    """返回 Agent prompt 文件的绝对路径。"""
    if agent_id not in AGENT_REGISTRY:
        raise ValueError(f"未知 Agent: {agent_id}，可用: {list(AGENT_REGISTRY)}")
    return AGENTS_DIR / AGENT_REGISTRY[agent_id]["path"]


def load_agent_prompt(agent_id: str) -> str:
    """加载 Agent prompt 原始内容。"""
    path = get_agent_path(agent_id)
    if not path.exists():
        raise FileNotFoundError(f"Agent prompt 不存在: {path}")
    return path.read_text(encoding="utf-8")


def list_zone_agents(zone: str) -> list[str]:
    """列出指定战区的所有 Agent ID。"""
    return sorted([
        agent_id for agent_id, info in AGENT_REGISTRY.items()
        if info["zone"] == zone
    ])


def list_all_agents() -> dict[str, list[str]]:
    """按战区分组列出所有 Agent。"""
    result: dict[str, list[str]] = {}
    for zone in ["prepare", "analyze", "design", "review", "monitor"]:
        result[zone] = list_zone_agents(zone)
    return result


def inject_context(prompt: str, context: dict[str, Any]) -> str:
    """向 prompt 注入运行时上下文变量。

    支持的占位符:
        {{BASE_NAME}}          → 需求基础名
        {{MANIFEST_DIR}}       → manifest 目录
        {{PREPARE_MANIFEST}}   → prepare manifest 路径
        {{ANALYZE_MANIFEST}}   → analyze manifest 路径
        {{DESIGN_MANIFEST}}    → design manifest 路径
        {{PROJECT_PROFILE}}    → 项目画像路径
        {{FLEET_CONFIG}}       → fleet_config.yml 路径
    """
    replacements = {
        "{{BASE_NAME}}": context.get("base_name", ""),
        "{{MANIFEST_DIR}}": str(REPO_ROOT / "output" / "manifests"),
        "{{PREPARE_MANIFEST}}": str(REPO_ROOT / "output" / "manifests" / f"{context.get('base_name', '')}_prepare.json"),
        "{{ANALYZE_MANIFEST}}": str(REPO_ROOT / "output" / "manifests" / f"{context.get('base_name', '')}_analyze.json"),
        "{{DESIGN_MANIFEST}}": str(REPO_ROOT / "output" / "manifests" / f"{context.get('base_name', '')}_design.json"),
        "{{PROJECT_PROFILE}}": str(REPO_ROOT / "knowledge_base" / "00_project" / "project_profile.md"),
        "{{FLEET_CONFIG}}": str(REPO_ROOT / "fleet_config.yml"),
        "{{OUTPUT_DIR}}": str(REPO_ROOT / "output"),
        "{{KNOWLEDGE_BASE_DIR}}": str(REPO_ROOT / "knowledge_base"),
        "{{REQUIREMENT_FILE}}": context.get("requirement_file", ""),
    }
    result = prompt
    for placeholder, value in replacements.items():
        result = result.replace(placeholder, value)
    return result


def render_agent_prompt(agent_id: str, context: dict[str, Any]) -> str:
    """加载并渲染 Agent prompt（注入上下文）。"""
    raw = load_agent_prompt(agent_id)
    return inject_context(raw, context)


def get_agent_frontmatter(agent_id: str) -> dict[str, Any]:
    """提取 Agent prompt 的 YAML frontmatter。"""
    content = load_agent_prompt(agent_id)
    frontmatter: dict[str, Any] = {}
    if content.startswith("---"):
        parts = content.split("---", 2)
        if len(parts) >= 3:
            for line in parts[1].strip().split("\n"):
                line = line.strip()
                if ":" in line:
                    key, value = line.split(":", 1)
                    frontmatter[key.strip()] = value.strip()
    return frontmatter


def get_agent_short_description(agent_id: str) -> str:
    """返回 Agent 的一句话描述。"""
    info = AGENT_REGISTRY.get(agent_id, {})
    return info.get("description", agent_id)


def validate_all_agents() -> dict[str, Any]:
    """验证所有 Agent prompt 文件是否存在且非空。"""
    result: dict[str, Any] = {"valid": True, "missing": [], "empty": [], "total": len(AGENT_REGISTRY)}
    for agent_id in AGENT_REGISTRY:
        path = get_agent_path(agent_id)
        if not path.exists():
            result["missing"].append(agent_id)
            result["valid"] = False
        elif path.stat().st_size == 0:
            result["empty"].append(agent_id)
            result["valid"] = False
    return result