Agent 审批系统：高风险命令、数据库迁移与生产发布门禁

2026-06-01 09:15  Agent 接到任务："优化数据库查询性能"
2026-06-01 09:16  Agent 分析慢查询，发现需要添加联合索引
2026-06-01 09:17  Agent 执行 ALTER TABLE orders ADD INDEX idx_user_date (user_id, created_at)
2026-06-01 09:17  索引构建锁表 45 秒，线上订单服务 503
2026-06-01 09:18  Agent 觉得索引不够，执行 DROP INDEX idx_old_status 删除旧索引
2026-06-01 09:18  旧索引被删除，依赖该索引的报表查询全部超时

                    Agent 发起命令
                         │
                         ▼
              ┌─────────────────────┐
              │   Command Proxy     │
              │   (命令分级)        │
              └──────────┬──────────┘
                         │
              ┌──────────▼──────────┐
              │  Risk Classification │
              │  SAFE → 直接执行     │
              │  MODERATE → 审计执行  │
              │  FORBIDDEN → 拒绝    │
              │  HIGH → 进入审批流程  │
              └──────────┬──────────┘
                         │ (HIGH)
                         ▼
              ┌─────────────────────┐
              │   Approval Engine   │
              │                     │
              │  1. 匹配审批策略     │
              │  2. 生成审批事件     │
              │  3. 通知审批人       │
              │  4. 暂停执行器       │
              └──────────┬──────────┘
                         │
              ┌──────────▼──────────┐
              │   Approval Gate     │
              │                     │
              │  approved → 恢复执行 │
              │  rejected → 拒绝+记录│
              │  timeout  → 按策略   │
              └─────────────────────┘

# approval-policies.yaml
policies:
  # 策略 1：数据库 DDL 操作需要 DBA 审批
  - name: "database-ddl"
    priority: 10
    match:
      command_patterns:
        - "ALTER TABLE*"
        - "DROP TABLE*"
        - "CREATE INDEX*"
        - "DROP INDEX*"
        - "npm run migrate*"
        - "python manage.py migrate*"
      risk_level: "high"
    approval:
      required_roles: ["dba", "tech-lead"]
      min_approvers: 1
      timeout_secs: 1800          # 30 分钟
      timeout_action: "reject"    # 超时视为拒绝
      scope: "per-command"        # 每条命令单独审批

  # 策略 2：生产环境部署需要运维审批
  - name: "production-deploy"
    priority: 20
    match:
      command_patterns:
        - "kubectl apply*"
        - "helm upgrade*"
        - "docker push*prod*"
        - "npm run deploy*"
      target_env: "production"
    approval:
      required_roles: ["devops", "sre"]
      min_approvers: 1
      timeout_secs: 3600          # 1 小时
      timeout_action: "reject"
      scope: "per-task"           # 同一任务的部署命令共享审批

  # 策略 3：Git 高风险操作需要 Tech Lead 审批
  - name: "git-dangerous"
    priority: 15
    match:
      command_patterns:
        - "git push --force*"
        - "git push*main"
        - "git push*master"
        - "git branch -D*"
        - "git reset --hard*"
    approval:
      required_roles: ["tech-lead"]
      min_approvers: 1
      timeout_secs: 600
      timeout_action: "reject"
      scope: "per-command"

  # 策略 4：敏感文件修改需要安全团队审批
  - name: "sensitive-files"
    priority: 25
    match:
      file_patterns:
        - "**/auth/**"
        - "**/crypto/**"
        - "**/.env*"
        - "**/secrets/**"
        - "**/Dockerfile"
    approval:
      required_roles: ["security"]
      min_approvers: 1
      timeout_secs: 1800
      timeout_action: "escalate"  # 超时升级到更高级别
      scope: "per-command"

  # 默认策略：未匹配的高风险命令
  - name: "default-high-risk"
    priority: 100
    match:
      risk_level: "high"
    approval:
      required_roles: ["tech-lead", "admin"]
      min_approvers: 1
      timeout_secs: 900
      timeout_action: "reject"
      scope: "per-command"

import fnmatch
from dataclasses import dataclass
from typing import Optional

@dataclass
class ApprovalRequest:
    """审批请求"""
    task_id: str
    command: str
    risk_level: str
    target_files: list[str]
    target_env: str
    requested_by: str
    agent_id: str

@dataclass
class ApprovalPolicy:
    """审批策略"""
    name: str
    priority: int
    match_rules: dict
    approval_config: dict

class PolicyEngine:
    """审批策略引擎"""

    def __init__(self, policies: list[ApprovalPolicy]):
        # 按优先级排序，数字小的先匹配
        self.policies = sorted(policies, key=lambda p: p.priority)

    def match(self, request: ApprovalRequest) -> Optional[ApprovalPolicy]:
        """找到第一个匹配的策略"""
        for policy in self.policies:
            if self._matches(policy, request):
                return policy
        return None

    def _matches(self, policy: ApprovalPolicy, req: ApprovalRequest) -> bool:
        rules = policy.match_rules

        # 命令模式匹配
        if "command_patterns" in rules:
            cmd_match = any(
                fnmatch.fnmatch(req.command, pat)
                for pat in rules["command_patterns"]
            )
            if not cmd_match:
                return False

        # 文件模式匹配
        if "file_patterns" in rules:
            file_match = any(
                fnmatch.fnmatch(f, pat)
                for f in req.target_files
                for pat in rules["file_patterns"]
            )
            if not file_match:
                return False

        # 风险等级匹配
        if "risk_level" in rules:
            if req.risk_level != rules["risk_level"]:
                return False

        # 目标环境匹配
        if "target_env" in rules:
            if req.target_env != rules["target_env"]:
                return False

        return True

class ApprovalSessionManager:
    """管理审批会话，支持 scope override"""

    def __init__(self, store):
        self.store = store

    def get_or_create_session(
        self, request: ApprovalRequest, policy: ApprovalPolicy
    ) -> str:
        """获取或创建审批会话，返回 session_id"""
        scope = policy.approval_config.get("scope", "per-command")

        if scope == "per-task":
            # 查找同一任务是否已有同策略的审批会话
            existing = self.store.find_session(
                task_id=request.task_id,
                policy_name=policy.name,
                status="active"
            )
            if existing:
                return existing.session_id

        # 创建新会话
        return self.store.create_session(
            task_id=request.task_id,
            policy_name=policy.name,
            command=request.command,
            requested_by=request.requested_by
        )

CREATE TABLE approval_events (
    id              UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    session_id      UUID NOT NULL,           -- 审批会话 ID
    task_id         UUID NOT NULL,           -- 关联任务 ID
    policy_name     VARCHAR(64) NOT NULL,    -- 匹配的策略名称
    -- 审批内容
    command         TEXT NOT NULL,           -- 请求执行的命令
    risk_level      VARCHAR(16) NOT NULL,    -- 风险等级
    context         JSONB,                   -- 附加上下文（diff、影响范围等）
    -- 审批流程
    status          VARCHAR(16) NOT NULL DEFAULT 'pending',
                    -- pending → approved / rejected / timeout / escalated
    requested_by    VARCHAR(64) NOT NULL,    -- 请求人（Agent ID 或用户）
    approved_by     VARCHAR(64),             -- 审批人
    approved_at     TIMESTAMPTZ,             -- 审批时间
    reason          TEXT,                    -- 审批意见（批准理由或拒绝原因）
    -- 超时控制
    timeout_at      TIMESTAMPTZ NOT NULL,    -- 超时时间
    timeout_action  VARCHAR(16) NOT NULL DEFAULT 'reject',
    -- 审计
    created_at      TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at      TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

-- 查询索引
CREATE INDEX idx_approval_task ON approval_events (task_id, status);
CREATE INDEX idx_approval_pending ON approval_events (status, timeout_at)
    WHERE status = 'pending';

# approval-roles.yaml
roles:
  tech-lead:
    can_approve:
      - "database-ddl"
      - "git-dangerous"
      - "default-high-risk"
    can_reject: true
    can_escalate: true

  dba:
    can_approve:
      - "database-ddl"
    can_reject: true
    can_escalate: false

  devops:
    can_approve:
      - "production-deploy"
    can_reject: true
    can_escalate: true

  sre:
    can_approve:
      - "production-deploy"
    can_reject: true
    can_escalate: true

  security:
    can_approve:
      - "sensitive-files"
    can_reject: true
    can_escalate: true

  admin:
    can_approve: ["*"]      # 可以审批所有策略
    can_reject: true
    can_escalate: true

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

class ApprovalDecision(BaseModel):
    decision: str          # "approved" | "rejected" | "escalated"
    approver: str          # 审批人 ID
    reason: str = ""       # 审批意见

@app.post("/api/approvals/{event_id}/decide")
async def submit_decision(event_id: str, decision: ApprovalDecision):
    """提交审批决策"""
    event = await store.get_event(event_id)
    if not event:
        raise HTTPException(404, "审批事件不存在")
    if event.status != "pending":
        raise HTTPException(409, f"审批已结束，当前状态: {event.status}")

    # 验证审批人权限
    policy = await get_policy(event.policy_name)
    role = await get_user_role(decision.approver)
    if not can_approve(role, policy.name):
        raise HTTPException(403, "无权审批此策略")

    # 更新审批事件
    await store.update_event(event_id, {
        "status": decision.decision,
        "approved_by": decision.approver,
        "approved_at": "NOW()",
        "reason": decision.reason,
    })

    # 通知执行器恢复或终止
    await notify_executor(event.task_id, decision.decision)

    return {"status": decision.decision, "event_id": event_id}

import asyncio
from datetime import datetime, timezone

class ApprovalTimeoutWatcher:
    """审批超时检测器"""

    def __init__(self, store, check_interval: int = 30):
        self.store = store
        self.check_interval = check_interval

    async def run(self):
        """后台循环检测超时"""
        while True:
            now = datetime.now(timezone.utc)
            expired = await self.store.find_expired(now)

            for event in expired:
                await self._handle_timeout(event)

            await asyncio.sleep(self.check_interval)

    async def _handle_timeout(self, event):
        action = event.timeout_action

        if action == "reject":
            await self.store.update_event(event.id, {
                "status": "timeout",
                "reason": f"审批超时（{event.timeout_at}），自动拒绝"
            })
            await notify_executor(event.task_id, "rejected")
            await notify_requester(event, "timeout_reject")

        elif action == "escalate":
            # 升级到 admin 角色
            await self.store.update_event(event.id, {
                "status": "pending",
                "policy_name": "escalated-to-admin",
                "timeout_at": "NOW() + interval '30 minutes'"
            })
            await notify_role("admin", event, "escalated")

        elif action == "notify_and_wait":
            # 再次通知，延长超时
            await self.store.update_event(event.id, {
                "timeout_at": "NOW() + interval '30 minutes'"
            })
            await notify_approvers(event, "reminder")

class ExecutorResumer:
    """审批后恢复执行器"""

    async def resume(self, task_id: str, decision: str, context: dict):
        """恢复被审批暂停的任务"""
        executor = await get_executor(task_id)

        if decision == "approved":
            # 将审批意见注入 Agent 上下文
            approval_context = {
                "type": "approval_result",
                "status": "approved",
                "approver": context["approver"],
                "reason": context.get("reason", ""),
                "additional_instructions": context.get("instructions", "")
            }

            # 如果审批人有附加指令，Agent 需要遵守
            # 例如："索引创建请在业务低峰期执行" → Agent 应等待或调整执行策略
            await executor.inject_context(approval_context)
            await executor.resume_command()

        elif decision == "rejected":
            # 拒绝时，Agent 需要知道原因并调整策略
            rejection_context = {
                "type": "approval_result",
                "status": "rejected",
                "reason": context.get("reason", "审批被拒绝"),
                "suggestion": context.get("suggestion", "")
            }
            await executor.inject_context(rejection_context)
            # Agent 收到拒绝后应该：
            # 1. 跳过当前命令，尝试替代方案
            # 2. 如果没有替代方案，标记任务失败
            await executor.skip_and_continue()

# notification-channels.yaml
notifications:
  pending_approval:
    channels:
      - type: slack
        target: "#approvals"
        template: |
          🔔 Agent 审批请求
          任务: {{task_title}}
          命令: {{command}}
          风险: {{risk_level}}
          策略: {{policy_name}}
          [批准]({{approve_url}}) | [拒绝]({{reject_url}})

      - type: email
        target: "{{approver_emails}}"
        subject: "[Agent 审批] {{policy_name}} - {{task_title}}"

      - type: webhook
        url: "https://internal-tools.example.com/agent-approvals"
        method: POST

  timeout_warning:
    # 超时前 5 分钟发提醒
    trigger_before_secs: 300
    channels:
      - type: slack
        target: "{{approver_dm}}"
        template: "⏰ 审批即将超时：{{command}}"

# 批量审批：审批人看到聚合后的完整变更
{
    "session_id": "sess-001",
    "task_title": "重构认证模块",
    "policy": "sensitive-files",
    "commands_count": 12,
    "files_affected": ["auth/login.py", "auth/session.py", ...],
    "full_diff": "...(完整 diff)...",
    "decision": "approved",
    "reason": "变更合理，测试覆盖充分"
}

{
    "decision": "rejected",
    "reason": "该索引被报表服务依赖，删除会导致报表超时",
    "suggestion": "请先添加替代索引后再删除旧索引，建议索引: idx_user_date_status",
    "constraints": [
        "不要删除 idx_old_status 索引",
        "新索引必须覆盖 user_id + created_at 的查询"
    ]
}

from datetime import time

def effective_timeout(request_time, timeout_secs, work_hours=(time(9, 0), time(18, 0))):
    """计算考虑工作时间的有效超时时间"""
    current = request_time.time()
    start, end = work_hours

    if current < start:
        # 工作开始前，从工作开始时间算起
        delta = (datetime.combine(request_time.date(), start) - request_time).total_seconds()
        return timeout_secs + delta
    elif current > end:
        # 工作结束后，从明天工作开始时间算起
        tomorrow = request_time.date() + timedelta(days=1)
        delta = (datetime.combine(tomorrow, start) - request_time).total_seconds()
        return timeout_secs + delta
    else:
        return timeout_secs