gaoandClaude
87a1ebcca5
- 创建Process-Driven AI Execution核心框架 - 设计四层契约体系 (需求/技术/执行/验证) - 实现Claude四种执行模式 (Design/Development/Review/Optimization) - 构建Process Gatekeeper三层验证体系 - 建立完整的质量门禁和违规处理机制 - 提供可执行的JSON Schema和模板库 核心价值： ✅ 一键创建 → Claude执行 → 自动验证 → 审阅 ✅ 将Claude从"代码助手"转变为"流程化工程执行体" ✅ 全程可控、可审、可追踪的AI执行管理 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <[email protected]>
726 lines
23 KiB
Markdown
726 lines
23 KiB
Markdown
# 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执行的质量和安全性。* |