# Process Gatekeeper（流程守卫系统）

> **版本**: v1.0
> **定位**: Claude执行流程的自动验证和质量控制中心
> **严格度**: ★★★★★ (所有规则必须严格执行)

---

## 🛡️ 系统架构

```mermaid
graph TD
    A[Claude执行开始] --> B[加载Execution Contract]
    B --> C[预执行检查]
    C --> D{预检查通过?}
    D -->|否| E[拒绝执行，返回错误]
    D -->|是| F[Claude执行任务]
    F --> G[生成输出结果]
    G --> H[实时违规检测]
    H --> I{发现违规?}
    I -->|是| J[触发违规处理]
    I -->|否| K[执行后验证]
    K --> L{验证通过?}
    L -->|否| M[退回重新执行]
    L -->|是| N[进入人工审核]
    N --> O[审核结果]
    O --> P{审核通过?}
    P -->|否| Q[反馈修改意见]
    P -->|是| R[任务完成]
    Q --> F
    J --> S[记录违规]
    S --> T[通知管理员]
    M --> F
```

---

## 🔍 验证层级

### 第一层：预执行检查（Pre-Execution Check）

#### 1.1 契约完整性验证
```python
def validate_execution_contract(contract: dict) -> ValidationResult:
    """验证执行契约的完整性和有效性"""

    required_fields = [
        'execution_id', 'phase', 'mode', 'requirement_id',
        'process_standard', 'quality_gates', 'deliverables',
        'constraints', 'verification'
    ]

    # 检查必需字段
    missing_fields = [field for field in required_fields if field not in contract]
    if missing_fields:
        return ValidationResult(
            valid=False,
            error_code="MISSING_REQUIRED_FIELDS",
            message=f"缺少必需字段: {missing_fields}"
        )

    # 验证格式规范
    if not re.match(r'^EXE-\d{4}-\d{4}$', contract['execution_id']):
        return ValidationResult(
            valid=False,
            error_code="INVALID_EXECUTION_ID",
            message="执行ID格式错误，应为 EXE-YYYY-NNNN"
        )

    # 验证阶段和模式一致性
    phase_mode_validations = {
        'requirement': ['design'],
        'design': ['design'],
        'development': ['development'],
        'testing': ['review'],
        'review': ['review', 'optimization']
    }

    if contract['mode'] not in phase_mode_validations.get(contract['phase'], []):
        return ValidationResult(
            valid=False,
            error_code="PHASE_MODE_MISMATCH",
            message=f"阶段 {contract['phase']} 与模式 {contract['mode']} 不匹配"
        )

    return ValidationResult(valid=True)
```

#### 1.2 权限验证
```python
def validate_permissions(contract: dict, user_context: dict) -> ValidationResult:
    """验证用户权限和操作范围"""

    # 检查用户是否有执行该阶段的权限
    user_role = user_context.get('role')
    phase = contract.get('phase')

    role_permissions = {
        'architect': ['requirement', 'design'],
        'developer': ['development'],
        'tester': ['testing'],
        'reviewer': ['review'],
        'admin': ['requirement', 'design', 'development', 'testing', 'review']
    }

    if phase not in role_permissions.get(user_role, []):
        return ValidationResult(
            valid=False,
            error_code="INSUFFICIENT_PERMISSIONS",
            message=f"角色 {user_role} 无权限执行 {phase} 阶段"
        )

    # 检查操作范围限制
    scope_limitations = contract.get('constraints', {}).get('scope_limitations', {})
    allowed_dirs = scope_limitations.get('allowed_directories', [])
    forbidden_dirs = scope_limitations.get('forbidden_directories', [])

    # TODO: 实现目录权限检查逻辑

    return ValidationResult(valid=True)
```

#### 1.3 资源可用性检查
```python
def check_resource_availability(contract: dict) -> ValidationResult:
    """检查执行所需资源是否可用"""

    # 检查相关文档是否存在
    context = contract.get('context', {})
    related_docs = context.get('related_docs', [])

    for doc_path in related_docs:
        if not os.path.exists(doc_path):
            return ValidationResult(
                valid=False,
                error_code="DOCUMENT_NOT_FOUND",
                message=f"相关文档不存在: {doc_path}"
            )

    # 检查质量门禁工具可用性
    required_gates = contract.get('quality_gates', {}).get('required_gates', [])

    gate_tools = {
        'lint': ['eslint', 'pylint'],
        'typecheck': ['tsc', 'mypy'],
        'unit_test': ['pytest', 'vitest'],
        'security_scan': ['bandit', 'trivy']
    }

    for gate in required_gates:
        tools = gate_tools.get(gate, [])
        if not any(tool_available(tool) for tool in tools):
            return ValidationResult(
                valid=False,
                error_code="QUALITY_GATE_UNAVAILABLE",
                message=f"质量门禁 {gate} 的工具不可用"
            )

    return ValidationResult(valid=True)
```

### 第二层：实时违规检测（Real-time Violation Detection）

#### 2.1 输出内容监控
```python
class ContentMonitor:
    """实时监控Claude输出内容"""

    def __init__(self, contract: dict):
        self.contract = contract
        self.forbidden_patterns = contract.get('constraints', {}).get('forbidden_patterns', [])
        self.required_patterns = contract.get('constraints', {}).get('required_patterns', [])
        self.forbidden_actions = contract.get('constraints', {}).get('forbidden_actions', [])

    def monitor_output(self, content: str, file_path: str) -> List[Violation]:
        """监控输出内容，检测违规行为"""
        violations = []

        # 检查禁止模式
        for pattern in self.forbidden_patterns:
            if re.search(pattern, content, re.IGNORECASE):
                violations.append(Violation(
                    type="FORBIDDEN_PATTERN",
                    severity="high",
                    pattern=pattern,
                    file_path=file_path,
                    message=f"检测到禁止模式: {pattern}"
                ))

        # 检查跨阶段操作
        phase = self.contract.get('phase')
        if phase == 'design' and self._contains_code(content):
            violations.append(Violation(
                type="CROSS_PHASE_OPERATION",
                severity="critical",
                message="设计阶段禁止生成代码"
            ))

        # 检查禁止操作
        for action in self.forbidden_actions:
            if action.lower() in content.lower():
                violations.append(Violation(
                    type="FORBIDDEN_ACTION",
                    severity="high",
                    action=action,
                    file_path=file_path,
                    message=f"检测到禁止操作: {action}"
                ))

        return violations

    def _contains_code(self, content: str) -> bool:
        """检测内容是否包含可执行代码"""
        code_indicators = [
            r'function\s+\w+\s*\(',
            r'class\s+\w+\s*:',
            r'def\s+\w+\s*\(',
            r'import\s+\w+',
            r'const\s+\w+\s*=',
            r'let\s+\w+\s*=',
            r'var\s+\w+\s*='
        ]

        return any(re.search(pattern, content) for pattern in code_indicators)
```

#### 2.2 交付物完整性检查
```python
class DeliverableTracker:
    """跟踪交付物生成状态"""

    def __init__(self, contract: dict):
        self.contract = contract
        self.expected_deliverables = contract.get('deliverables', [])
        self.generated_deliverables = {}

    def track_deliverable(self, file_path: str, content: str) -> ValidationResult:
        """跟踪交付物生成"""

        # 查找对应的交付物定义
        expected = None
        for deliverable in self.expected_deliverables:
            if deliverable['path'] == file_path:
                expected = deliverable
                break

        if not expected:
            return ValidationResult(
                valid=False,
                error_code="UNEXPECTED_DELIVERABLE",
                message=f"未预期的交付物: {file_path}"
            )

        # 验证内容类型
        expected_type = expected.get('type')
        if not self._validate_content_type(content, expected_type):
            return ValidationResult(
                valid=False,
                error_code="INVALID_CONTENT_TYPE",
                message=f"交付物 {file_path} 内容类型与预期不符"
            )

        # 验证Schema约束
        schema = expected.get('schema')
        if schema and not self._validate_schema(content, schema):
            return ValidationResult(
                valid=False,
                error_code="SCHEMA_VALIDATION_FAILED",
                message=f"交付物 {file_path} 不符合Schema约束"
            )

        self.generated_deliverables[file_path] = {
            'content': content,
            'type': expected_type,
            'size': len(content),
            'created_at': datetime.now().isoformat()
        }

        return ValidationResult(valid=True)

    def check_completeness(self) -> ValidationResult:
        """检查交付物完整性"""

        missing_deliverables = []
        for expected in self.expected_deliverables:
            if expected['path'] not in self.generated_deliverables:
                missing_deliverables.append(expected['path'])

        if missing_deliverables:
            return ValidationResult(
                valid=False,
                error_code="INCOMPLETE_DELIVERABLES",
                message=f"缺少交付物: {missing_deliverables}"
            )

        return ValidationResult(valid=True)
```

### 第三层：执行后验证（Post-Execution Validation）

#### 3.1 质量门禁执行
```python
class QualityGateExecutor:
    """执行质量门禁检查"""

    def __init__(self, contract: dict):
        self.contract = contract
        self.quality_gates = contract.get('quality_gates', {})
        self.required_gates = self.quality_gates.get('required_gates', [])
        self.thresholds = self.quality_gates.get('thresholds', {})

    async def execute_quality_gates(self, workspace_path: str) -> QualityGateResult:
        """执行所有质量门禁检查"""

        results = {}
        overall_passed = True

        for gate in self.required_gates:
            try:
                if gate == 'lint':
                    result = await self._run_lint_check(workspace_path)
                elif gate == 'typecheck':
                    result = await self._run_typecheck(workspace_path)
                elif gate == 'unit_test':
                    result = await self._run_unit_tests(workspace_path)
                elif gate == 'security_scan':
                    result = await self._run_security_scan(workspace_path)
                elif gate == 'schema_validation':
                    result = await self._run_schema_validation(workspace_path)
                else:
                    result = QualityGateResult(
                        gate=gate,
                        passed=False,
                        message=f"未知的质量门禁: {gate}"
                    )

                results[gate] = result

                if not result.passed:
                    overall_passed = False

            except Exception as e:
                results[gate] = QualityGateResult(
                    gate=gate,
                    passed=False,
                    message=f"执行质量门禁失败: {str(e)}"
                )
                overall_passed = False

        # 计算质量分数
        quality_score = self._calculate_quality_score(results)

        return QualityGateResult(
            gate="overall",
            passed=overall_passed and quality_score >= self.thresholds.get('quality_score', 80),
            message=f"整体质量分数: {quality_score}",
            details=results,
            quality_score=quality_score
        )

    async def _run_lint_check(self, workspace_path: str) -> QualityGateResult:
        """运行代码规范检查"""

        # 检测项目类型和选择合适的linter
        if os.path.exists(os.path.join(workspace_path, 'package.json')):
            return await self._run_eslint(workspace_path)
        elif os.path.exists(os.path.join(workspace_path, 'pyproject.toml')):
            return await self._run_pylint(workspace_path)
        else:
            return QualityGateResult(
                gate="lint",
                passed=False,
                message="无法确定项目类型，跳过lint检查"
            )

    async def _run_unit_tests(self, workspace_path: str) -> QualityGateResult:
        """运行单元测试"""

        # 执行测试并收集覆盖率
        try:
            # 运行测试
            test_result = subprocess.run(
                ['pytest', '--cov=.', '--cov-report=json'],
                cwd=workspace_path,
                capture_output=True,
                text=True
            )

            if test_result.returncode != 0:
                return QualityGateResult(
                    gate="unit_test",
                    passed=False,
                    message=f"测试失败: {test_result.stderr}"
                )

            # 读取覆盖率报告
            coverage_data = self._parse_coverage_report(workspace_path)
            coverage_percentage = coverage_data.get('totals', {}).get('percent_covered', 0)

            required_coverage = self.thresholds.get('test_coverage', 80)
            passed = coverage_percentage >= required_coverage

            return QualityGateResult(
                gate="unit_test",
                passed=passed,
                message=f"测试覆盖率: {coverage_percentage}% (要求: {required_coverage}%)",
                metrics={'coverage_percentage': coverage_percentage}
            )

        except Exception as e:
            return QualityGateResult(
                gate="unit_test",
                passed=False,
                message=f"执行测试失败: {str(e)}"
            )
```

#### 3.2 安全检查
```python
class SecurityScanner:
    """安全扫描器"""

    def __init__(self):
        self.security_rules = [
            self._check_hardcoded_secrets,
            self._check_sql_injection,
            self._check_xss_vulnerabilities,
            self._check_insecure_dependencies,
            self._check_file_permissions
        ]

    async def scan_security(self, workspace_path: str) -> SecurityResult:
        """执行安全扫描"""

        vulnerabilities = []
        severity_counts = {'critical': 0, 'high': 0, 'medium': 0, 'low': 0}

        for rule in self.security_rules:
            try:
                rule_vulnerabilities = await rule(workspace_path)
                vulnerabilities.extend(rule_vulnerabilities)

                for vuln in rule_vulnerabilities:
                    severity_counts[vuln.severity] += 1

            except Exception as e:
                logging.error(f"安全规则执行失败: {str(e)}")

        # 计算风险等级
        risk_level = self._calculate_risk_level(severity_counts)

        return SecurityResult(
            passed=severity_counts['critical'] == 0 and severity_counts['high'] == 0,
            risk_level=risk_level,
            vulnerabilities=vulnerabilities,
            summary=severity_counts
        )

    async def _check_hardcoded_secrets(self, workspace_path: str) -> List[Vulnerability]:
        """检查硬编码密钥"""

        secret_patterns = [
            (r'password\s*=\s*["\'][^"\']+["\']', 'hardcoded_password'),
            (r'api_key\s*=\s*["\'][^"\']+["\']', 'hardcoded_api_key'),
            (r'secret\s*=\s*["\'][^"\']+["\']', 'hardcoded_secret'),
            (r'token\s*=\s*["\'][^"\']+["\']', 'hardcoded_token')
        ]

        vulnerabilities = []

        for root, dirs, files in os.walk(workspace_path):
            # 跳过常见的非源码目录
            dirs[:] = [d for d in dirs if d not in ['.git', 'node_modules', '__pycache__', '.venv']]

            for file in files:
                if file.endswith(('.py', '.js', '.ts', '.java', '.go', '.rs')):
                    file_path = os.path.join(root, file)

                    try:
                        with open(file_path, 'r', encoding='utf-8') as f:
                            content = f.read()

                        for pattern, vuln_type in secret_patterns:
                            matches = re.finditer(pattern, content, re.IGNORECASE)
                            for match in matches:
                                vulnerabilities.append(Vulnerability(
                                    type=vuln_type,
                                    severity="high",
                                    file_path=file_path,
                                    line_number=content[:match.start()].count('\n') + 1,
                                    message=f"发现硬编码密钥: {vuln_type}",
                                    recommendation="使用环境变量或配置管理工具存储敏感信息"
                                ))

                    except Exception as e:
                        logging.warning(f"无法扫描文件 {file_path}: {str(e)}")

        return vulnerabilities
```

---

## 🚨 违规处理机制

### 违规等级定义

| 等级 | 描述 | 处理方式 | 自动恢复 |
|------|------|----------|----------|
| **Critical** | 严重违规，如安全漏洞、跨阶段操作 | 立即停止执行，上报管理员 | 否 |
| **High** | 高风险违规，如违反核心约束 | 停止执行，需要人工干预 | 否 |
| **Medium** | 中等风险，如质量问题 | 记录违规，允许重试 | 是 |
| **Low** | 低风险，如格式问题 | 警告记录，继续执行 | 是 |

### 自动处理流程

```python
class ViolationHandler:
    """违规处理器"""

    def __init__(self, notification_service: NotificationService):
        self.notification_service = notification_service

    async def handle_violation(self, violation: Violation, context: ExecutionContext) -> HandlingResult:
        """处理违规事件"""

        # 记录违规
        await self._log_violation(violation, context)

        # 根据严重程度选择处理策略
        if violation.severity == 'critical':
            return await self._handle_critical_violation(violation, context)
        elif violation.severity == 'high':
            return await self._handle_high_violation(violation, context)
        elif violation.severity == 'medium':
            return await self._handle_medium_violation(violation, context)
        else:
            return await self._handle_low_violation(violation, context)

    async def _handle_critical_violation(self, violation: Violation, context: ExecutionContext) -> HandlingResult:
        """处理严重违规"""

        # 立即停止执行
        await context.stop_execution()

        # 清理已生成的文件
        await context.cleanup_workspace()

        # 通知管理员
        await self.notification_service.send_alert(
            level="critical",
            message=f"严重违规: {violation.message}",
            context=context.to_dict()
        )

        # 启动安全事故流程
        await self._initiate_security_incident(violation, context)

        return HandlingResult(
            action="execution_stopped",
            requires_manual_intervention=True,
            message="执行已因严重违规而停止，等待管理员处理"
        )

    async def _handle_medium_violation(self, violation: Violation, context: ExecutionContext) -> HandlingResult:
        """处理中等风险违规"""

        # 尝试自动修复
        if await self._attempt_auto_fix(violation, context):
            return HandlingResult(
                action="auto_fixed",
                requires_manual_intervention=False,
                message="违规已自动修复，继续执行"
            )

        # 记录违规并允许重试
        await self._record_warning(violation, context)

        return HandlingResult(
            action="warning_recorded",
            requires_manual_intervention=False,
            message="记录违规警告，允许重试"
        )
```

---

## 📊 监控和报告

### 实时监控指标

```python
class GatekeeperMetrics:
    """守卫系统监控指标"""

    def __init__(self):
        self.execution_count = Counter('gatekeeper_executions_total')
        self.violation_count = Counter('gatekeeper_violations_total', ['severity', 'type'])
        self.quality_gate_results = Histogram('gatekeeper_quality_gate_duration')
        self.success_rate = Gauge('gatekeeper_success_rate')

    def record_execution_start(self, execution_id: str, phase: str, mode: str):
        """记录执行开始"""
        self.execution_count.labels(phase=phase, mode=mode).inc()

    def record_violation(self, violation: Violation):
        """记录违规"""
        self.violation_count.labels(
            severity=violation.severity,
            type=violation.type
        ).inc()

    def record_quality_gate_result(self, gate: str, duration: float, passed: bool):
        """记录质量门禁结果"""
        self.quality_gate_results.labels(gate=gate).observe(duration)

    def calculate_success_rate(self, time_window: int = 3600) -> float:
        """计算成功率"""
        # 实现成功率计算逻辑
        pass
```

### 生成执行报告

```python
class ExecutionReporter:
    """执行报告生成器"""

    async def generate_execution_report(self, execution_id: str) -> ExecutionReport:
        """生成详细的执行报告"""

        # 收集执行数据
        execution_data = await self._collect_execution_data(execution_id)

        # 生成质量评估
        quality_assessment = await self._assess_quality(execution_data)

        # 分析违规情况
        violation_analysis = await self._analyze_violations(execution_data)

        # 计算效率指标
        efficiency_metrics = await self._calculate_efficiency(execution_data)

        return ExecutionReport(
            execution_id=execution_id,
            summary=execution_data.get('summary'),
            quality_assessment=quality_assessment,
            violation_analysis=violation_analysis,
            efficiency_metrics=efficiency_metrics,
            recommendations=await self._generate_recommendations(execution_data),
            generated_at=datetime.now().isoformat()
        )
```

---

## 🔧 配置和部署

### 系统配置

```yaml
# gatekeeper-config.yaml
gatekeeper:
  enabled: true
  strict_mode: true

  validation:
    pre_execution_checks:
      - contract_validation
      - permission_check
      - resource_availability

    real_time_monitoring:
      - content_monitoring
      - deliverable_tracking
      - violation_detection

    post_execution_validation:
      - quality_gates
      - security_scan
      - completeness_check

  violations:
    critical:
      auto_stop: true
      notification: ["admin", "security_team"]

    high:
      retry_limit: 1
      notification: ["admin"]

    medium:
      retry_limit: 3
      auto_fix: true

    low:
      log_only: true

  quality_gates:
    lint:
      enabled: true
      tools: ["eslint", "pylint", "gofmt"]

    typecheck:
      enabled: true
      tools: ["tsc", "mypy", "go build"]

    unit_test:
      enabled: true
      min_coverage: 80
      tools: ["pytest", "vitest", "go test"]

    security_scan:
      enabled: true
      tools: ["bandit", "trivy", "snyk"]
      fail_on_high: true
```

---

**更新记录**:

| 版本 | 日期 | 更新内容 | 更新人 |
|------|------|----------|--------|
| v1.0 | 2025-10-29 | 初始版本，定义完整守卫系统 | Claude AI |

*Process Gatekeeper 是确保 Claude 严格执行流程标准的核心组件，通过多层次验证和实时监控，保证AI执行的质量和安全性。*