Agent 质量门禁：测试、Lint、安全扫描与代码审查闭环

任务：为 UserService 添加批量删除功能

Agent 产出：
  ✅ 实现了 batchDelete() 方法
  ✅ 添加了 API 路由 /api/users/batch-delete
  ❌ 没有写单元测试
  ❌ 没有做输入验证（可以传入空数组、负数 ID）
  ❌ 没有检查权限（普通用户可以删除其他用户）
  ❌ 硬编码了数据库查询（SQL 注入风险）
  ❌ 没有更新 API 文档
  ❌ 代码风格与项目不一致（camelCase vs snake_case 混用）

Agent 产出代码
    │
    ▼
┌──────────────────────────────────────────────────────────────┐
│                     Quality Gate Pipeline                     │
│                                                               │
│  Gate 1          Gate 2          Gate 3          Gate 4      │
│  ┌─────────┐    ┌─────────┐    ┌─────────┐    ┌─────────┐  │
│  │  编译    │ →  │  测试    │ →  │  Lint   │ →  │  安全   │  │
│  │  检查    │    │  运行    │    │  检查    │    │  扫描   │  │
│  └────┬────┘    └────┬────┘    └────┬────┘    └────┬────┘  │
│       │              │              │              │          │
│  fail → 回退    fail → 回退    fail → 回退    fail → 回退    │
│                                                               │
│  Gate 5                                                       │
│  ┌─────────┐                                                  │
│  │  Review │ → 通过 → 创建 PR                                  │
│  │  摘要   │         失败 → 回退给 Agent 修复                   │
│  └─────────┘                                                  │
└──────────────────────────────────────────────────────────────┘

# quality-gates.yaml
gates:
  compile:
    enabled: true
    command: "npm run build"
    timeout_secs: 120
    blocking: true              # 阻塞级：失败立即终止

  unit_test:
    enabled: true
    command: "npm test -- --coverage --reporter=json"
    timeout_secs: 300
    blocking: true
    coverage:
      enabled: true
      min_line_coverage: 80     # 最低行覆盖率
      min_branch_coverage: 70   # 最低分支覆盖率
      # 只检查 Agent 修改的文件
      changed_files_only: true
    # Agent 修改了哪些文件，就运行对应的测试
    smart_test:
      enabled: true
      # 文件到测试的映射规则
      test_patterns:
        - source: "src/**/*.ts"
          test: "src/**/*.test.ts"
        - source: "src/**/*.py"
          test: "tests/**/test_*.py"

  integration_test:
    enabled: true
    command: "npm run test:integration"
    timeout_secs: 600
    blocking: false             # 非阻塞：失败记录但不终止
    # 只在特定目录有改动时运行
    trigger_paths:
      - "src/api/**"
      - "src/db/**"
      - "src/services/**"

import subprocess
import json
from pathlib import Path

class SmartTestSelector:
    """智能测试选择器：只运行与变更文件相关的测试"""

    def __init__(self, repo_root: str, config: dict):
        self.repo_root = Path(repo_root)
        self.test_patterns = config.get("test_patterns", [])

    def select_tests(self, changed_files: list[str]) -> list[str]:
        """根据变更文件选择需要运行的测试"""
        tests = set()

        for changed in changed_files:
            for pattern in self.test_patterns:
                if self._matches(changed, pattern["source"]):
                    # 找到对应的测试文件
                    test_glob = pattern["test"]
                    # 从源文件路径推导测试文件路径
                    test_files = self._derive_test_files(changed, test_glob)
                    tests.update(test_files)

        return sorted(tests)

    def run_selected_tests(self, tests: list[str], timeout: int = 300) -> dict:
        """运行选中的测试，返回结构化结果"""
        if not tests:
            return {"status": "skipped", "reason": "no relevant tests"}

        cmd = ["npx", "jest", "--json", "--testPathPattern",
               "|".join(tests)]

        result = subprocess.run(
            cmd, cwd=self.repo_root,
            capture_output=True, text=True,
            timeout=timeout
        )

        try:
            report = json.loads(result.stdout)
        except json.JSONDecodeError:
            return {"status": "error", "output": result.stdout + result.stderr}

        return {
            "status": "passed" if report.get("success") else "failed",
            "total": report.get("numTotalTests", 0),
            "passed": report.get("numPassedTests", 0),
            "failed": report.get("numFailedTests", 0),
            "failed_tests": [
                {
                    "name": t["fullName"],
                    "message": t["failureMessages"][0] if t.get("failureMessages") else ""
                }
                for suite in report.get("testResults", [])
                for t in suite.get("assertionResults", [])
                if t.get("status") == "failed"
            ],
            "coverage": self._extract_coverage(report),
        }

    def _derive_test_files(self, source: str, test_glob: str) -> list[str]:
        stem = Path(source).stem
        directory = Path(source).parent
        # 在同目录和 tests 目录中查找测试文件
        candidates = [
            str(directory / f"{stem}.test.ts"),
            str(directory / f"{stem}.spec.ts"),
            str(directory / f"test_{stem}.py"),
            str(directory / f"{stem}_test.go"),
            f"tests/**/test_{stem}*.py",
        ]
        found = []
        for c in candidates:
            matches = list(self.repo_root.glob(c))
            found.extend(str(m.relative_to(self.repo_root)) for m in matches)
        return found

    def _matches(self, path: str, pattern: str) -> bool:
        from fnmatch import fnmatch
        return fnmatch(path, pattern)

    def _extract_coverage(self, report: dict) -> dict:
        cov = report.get("coverageMap", {})
        if not cov:
            return {}
        total_lines = sum(v.get("lines", {}).get("total", 0) for v in cov.values())
        covered = sum(v.get("lines", {}).get("covered", 0) for v in cov.values())
        return {
            "line_coverage": (covered / total_lines * 100) if total_lines else 0,
            "files_covered": len(cov),
        }

  lint:
    enabled: true
    tools:
      - name: "eslint"
        command: "npx eslint --format json {changed_files}"
        blocking_level: "error"    # 只有 error 阻塞，warning 记录
      - name: "ruff"
        command: "ruff check --output-format json {changed_files}"
        blocking_level: "error"
      - name: "prettier"
        command: "npx prettier --check {changed_files}"
        blocking_level: "warning"  # 格式问题不阻塞
    # Agent 可以尝试自动修复 Lint 错误
    auto_fix:
      enabled: true
      max_fix_rounds: 2           # 最多尝试 2 轮自动修复
      fix_commands:
        - "npx eslint --fix {changed_files}"
        - "ruff check --fix {changed_files}"
        - "npx prettier --write {changed_files}"

import subprocess
import json
from dataclasses import dataclass

@dataclass
class SecurityFinding:
    severity: str         # critical, high, medium, low
    category: str         # sqli, xss, hardcoded-secret, dependency-vuln
    file: str
    line: int
    message: str
    rule_id: str

class SecurityScanner:
    """安全扫描门禁"""

    def __init__(self, repo_root: str, config: dict):
        self.repo_root = repo_root
        self.blocking_severity = config.get("blocking_severity", ["critical", "high"])

    def scan_all(self, changed_files: list[str]) -> dict:
        """运行所有安全扫描"""
        findings = []

        # 1. 依赖漏洞扫描
        findings.extend(self._scan_dependencies())

        # 2. 代码安全扫描（Semgrep）
        findings.extend(self._scan_code_semgrep(changed_files))

        # 3. 密钥泄露检测
        findings.extend(self._scan_secrets(changed_files))

        # 判断是否阻塞
        blocking = [f for f in findings if f.severity in self.blocking_severity]

        return {
            "status": "failed" if blocking else "passed",
            "total_findings": len(findings),
            "blocking_findings": len(blocking),
            "findings": [
                {
                    "severity": f.severity,
                    "category": f.category,
                    "file": f.file,
                    "line": f.line,
                    "message": f.message,
                    "rule_id": f.rule_id,
                }
                for f in findings
            ],
        }

    def _scan_dependencies(self) -> list[SecurityFinding]:
        """依赖漏洞扫描：npm audit / pip-audit"""
        findings = []
        try:
            result = subprocess.run(
                ["npm", "audit", "--json"],
                cwd=self.repo_root, capture_output=True, text=True, timeout=60
            )
            report = json.loads(result.stdout) if result.stdout else {}
            for vuln in report.get("vulnerabilities", {}).values():
                findings.append(SecurityFinding(
                    severity=vuln.get("severity", "medium"),
                    category="dependency-vuln",
                    file="package.json",
                    line=0,
                    message=f"{vuln['name']}@{vuln['range']}: {vuln.get('title', '')}",
                    rule_id=f"npm-audit-{vuln['name']}",
                ))
        except (subprocess.TimeoutExpired, json.JSONDecodeError, FileNotFoundError):
            pass
        return findings

    def _scan_code_semgrep(self, changed_files: list[str]) -> list[SecurityFinding]:
        """代码安全扫描：使用 Semgrep"""
        findings = []
        try:
            cmd = ["semgrep", "--json", "--config", "auto"] + changed_files
            result = subprocess.run(
                cmd, cwd=self.repo_root, capture_output=True, text=True, timeout=120
            )
            report = json.loads(result.stdout) if result.stdout else {}
            for finding in report.get("results", []):
                findings.append(SecurityFinding(
                    severity=finding.get("extra", {}).get("severity", "medium"),
                    category=finding.get("check_id", "unknown"),
                    file=finding.get("path", ""),
                    line=finding.get("start", {}).get("line", 0),
                    message=finding.get("extra", {}).get("message", ""),
                    rule_id=finding.get("check_id", ""),
                ))
        except (subprocess.TimeoutExpired, FileNotFoundError):
            pass
        return findings

    def _scan_secrets(self, changed_files: list[str]) -> list[SecurityFinding]:
        """密钥泄露检测"""
        findings = []
        try:
            cmd = ["gitleaks", "detect", "--no-git", "--report-format", "json",
                   "--report-path", "/tmp/gitleaks-report.json"]
            subprocess.run(cmd, cwd=self.repo_root, timeout=60)
        except (subprocess.TimeoutExpired, FileNotFoundError):
            pass
        return findings

class ReviewSummaryGenerator:
    """为人类 Reviewer 生成审查摘要"""

    def generate(self, task_id: str, diff: str, gate_results: dict) -> str:
        """生成 Markdown 格式的 Review 摘要"""
        lines = [
            f"## Agent 任务审查摘要 (Task #{task_id})",
            "",
            "### 变更概览",
            f"- 修改文件数: {gate_results.get('files_changed', 'N/A')}",
            f"- 新增行数: {gate_results.get('lines_added', 'N/A')}",
            f"- 删除行数: {gate_results.get('lines_removed', 'N/A')}",
            "",
            "### 门禁结果",
        ]

        for gate_name, result in gate_results.get("gates", {}).items():
            status = "✅" if result["status"] == "passed" else "❌"
            lines.append(f"- {status} **{gate_name}**: {result.get('detail', '')}")

        lines.extend([
            "",
            "### 需要人工审查的要点",
            "- [ ] 业务逻辑是否正确",
            "- [ ] 架构决策是否合理",
            "- [ ] 边界条件是否覆盖",
            "- [ ] 是否有过度工程化",
        ])

        # 附加安全扫描发现
        security = gate_results.get("gates", {}).get("security", {})
        if security.get("total_findings", 0) > 0:
            lines.extend([
                "",
                f"### 安全扫描发现 ({security['total_findings']} 项)",
            ])
            for f in security.get("findings", [])[:10]:
                lines.append(f"- [{f['severity']}] {f['file']}:{f['line']} - {f['message']}")

        return "\n".join(lines)

# .github/workflows/agent-quality-gate.yml
name: Agent Quality Gate
on:
  push:
    branches: ["agent/task-*"]

jobs:
  quality-gate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Install dependencies
        run: npm ci

      - name: Gate 1 - Build check
        run: npm run build
        timeout-minutes: 2

      - name: Gate 2 - Unit tests
        run: npm test -- --coverage --reporter=json --outputFile=test-report.json
        timeout-minutes: 5

      - name: Gate 3 - Lint check
        run: |
          npx eslint --format json $(git diff --name-only origin/main...HEAD -- '*.ts' '*.js')
        timeout-minutes: 3
        continue-on-error: true

      - name: Gate 4 - Security scan
        run: |
          npm audit --json > audit-report.json || true
          npx semgrep --json --config auto $(git diff --name-only origin/main...HEAD) > semgrep-report.json || true
        timeout-minutes: 5

      - name: Gate 5 - Generate review summary
        if: always()
        run: |
          python scripts/generate-review-summary.py \
            --test-report test-report.json \
            --audit-report audit-report.json \
            --semgrep-report semgrep-report.json \
            --output summary.md

      - name: Post summary to PR
        if: always()
        uses: actions/github-script@v7
        with:
          script: |
            const fs = require('fs');
            const summary = fs.readFileSync('summary.md', 'utf8');
            await github.rest.issues.createComment({
              owner: context.repo.owner,
              repo: context.repo.repo,
              issue_number: context.issue.number,
              body: summary
            });

import asyncio
from dataclasses import dataclass
from enum import Enum

class GateStatus(Enum):
    PASSED = "passed"
    FAILED = "failed"
    SKIPPED = "skipped"
    WARNING = "warning"

@dataclass
class GateResult:
    name: str
    status: GateStatus
    detail: str
    duration_secs: float
    auto_fixable: bool = False

class QualityGatePipeline:
    """质量门禁流水线"""

    def __init__(self, config: dict):
        self.config = config
        self.results: list[GateResult] = []

    async def run(self, changed_files: list[str]) -> dict:
        """运行整条门禁流水线"""
        gates = [
            ("compile", self._run_compile),
            ("unit_test", self._run_unit_tests),
            ("lint", self._run_lint),
            ("security", self._run_security),
            ("review_summary", self._generate_review),
        ]

        for gate_name, gate_func in gates:
            gate_config = self.config.get("gates", {}).get(gate_name, {})
            if not gate_config.get("enabled", True):
                self.results.append(GateResult(gate_name, GateStatus.SKIPPED, "disabled", 0))
                continue

            result = await gate_func(changed_files, gate_config)
            self.results.append(result)

            # 阻塞级门禁失败 → 终止流水线
            if result.status == GateStatus.FAILED and gate_config.get("blocking", True):
                # 尝试自动修复
                if result.auto_fixable and gate_config.get("auto_fix", {}).get("enabled"):
                    fixed = await self._try_auto_fix(gate_name, changed_files, gate_config)
                    if fixed:
                        # 重新运行门禁
                        result = await gate_func(changed_files, gate_config)
                        self.results[-1] = result

                if result.status == GateStatus.FAILED:
                    break

        return self._build_report()

    async def _try_auto_fix(self, gate_name: str, files: list[str], config: dict) -> bool:
        """尝试自动修复门禁失败"""
        max_rounds = config.get("auto_fix", {}).get("max_fix_rounds", 2)
        fix_commands = config.get("auto_fix", {}).get("fix_commands", [])

        for round_num in range(max_rounds):
            for cmd in fix_commands:
                formatted = cmd.replace("{changed_files}", " ".join(files))
                proc = await asyncio.create_subprocess_shell(
                    formatted, cwd=self.config.get("repo_root", ".")
                )
                await proc.wait()

            # 重新检查
            return True  # 让调用者重新运行门禁

        return False

    def _build_report(self) -> dict:
        """构建流水线报告"""
        all_passed = all(r.status in (GateStatus.PASSED, GateStatus.SKIPPED, GateStatus.WARNING)
                        for r in self.results)
        return {
            "status": "passed" if all_passed else "failed",
            "gates": {r.name: {
                "status": r.status.value,
                "detail": r.detail,
                "duration_secs": r.duration_secs,
            } for r in self.results},
            "total_duration": sum(r.duration_secs for r in self.results),
        }

# failure-strategy.yaml
on_gate_failure:
  # 策略 1：自动修复（适用于 Lint、格式化问题）
  auto_fix:
    enabled: true
    applicable_gates: ["lint", "compile"]
    max_rounds: 2
    # 修复后重新运行门禁

  # 策略 2：回退给 Agent（适用于测试失败）
  agent_retry:
    enabled: true
    applicable_gates: ["unit_test"]
    max_retries: 2
    # 把测试失败的错误信息注入 Agent 上下文，让它修复
    inject_failure_context: true
    context_template: |
      以下测试用例失败了，请修复：
      {{#each failed_tests}}
      - {{name}}: {{message}}
      {{/each}}

  # 策略 3：标记人工介入（适用于安全问题）
  human_intervention:
    enabled: true
    applicable_gates: ["security"]
    # 通知团队安全审查
    notify:
      channel: "#security-alerts"
      template: "Agent 产出的代码存在安全问题，需要人工审查"

  # 策略 4：放弃任务（多次修复都失败）
  abort:
    trigger: "all_retries_exhausted"
    # 标记任务失败，保留现场
    keep_artifacts: true

# 错误做法：只看覆盖率
if coverage < 80:
    return GateResult.FAILED

# 正确做法：检查测试是否存在且有意义的断言
def check_test_quality(test_file: str) -> bool:
    content = Path(test_file).read_text()
    # 检查是否有真正的断言（不是 toBeDefined）
    meaningful_asserts = re.findall(
        r'expect\(.*\)\.(toBe|toEqual|toContain|toThrow|toHaveProperty)',
        content
    )
    return len(meaningful_asserts) >= 2  # 至少 2 个有意义的断言

security:
  scan_paths:
    include: ["src/**"]
    exclude:
      - "src/**/*.test.*"
      - "src/**/*.spec.*"
      - "tests/**"
      - "**/__mocks__/**"
      - "**/*.fixture.*"
  suppress_rules:
    - id: "hardcoded-secret"
      paths: ["**/*.test.*", "**/fixtures/**"]
      reason: "测试中的假密钥不是真实漏洞"

def select_gates(changed_files: list[str]) -> list[str]:
    """根据变更文件选择需要运行的门禁"""
    gates = ["compile"]  # 编译检查永远运行

    has_code = any(f.endswith(('.ts', '.js', '.py', '.go')) for f in changed_files)
    has_deps = any(f in ('package.json', 'package-lock.json', 'requirements.txt') for f in changed_files)
    has_config = any(f.endswith(('.yaml', '.yml', '.json', '.toml')) for f in changed_files)

    if has_code:
        gates.extend(["unit_test", "lint", "security"])
    if has_deps:
        gates.append("dependency_audit")
    if has_code or has_config:
        gates.append("review_summary")

    return gates