Agent 审计日志规范：工具调用、Prompt、diff 与审批记录

-- Agent 审计日志主表
CREATE TABLE agent_audit_logs (
    id              BIGSERIAL PRIMARY KEY,
    event_time      TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    
    -- 会话信息
    session_id      VARCHAR(64) NOT NULL,
    task_id         VARCHAR(64),
    
    -- 操作者信息
    user_id         VARCHAR(64) NOT NULL,
    user_name       VARCHAR(128),
    agent_type      VARCHAR(32),           -- claude-code / codex / cursor / ...
    model_id        VARCHAR(64),
    
    -- 事件类型
    event_type      VARCHAR(32) NOT NULL,  -- tool_call / prompt / diff / approval / cost
    event_action    VARCHAR(64),           -- read_file / write_file / run_command / ...
    
    -- 事件内容（JSON）
    payload         JSONB NOT NULL,
    
    -- 元信息
    risk_level      VARCHAR(16),           -- LOW / MEDIUM / HIGH / CRITICAL
    approval_id     VARCHAR(64),           -- 关联的审批记录
    cost_usd        DECIMAL(10,6),
    tokens_used     INTEGER,
    
    -- 索引字段
    project_id      VARCHAR(64),
    repo_name       VARCHAR(256),
    
    -- 完整性
    checksum        VARCHAR(64)            -- SHA-256 校验和（防篡改）
);

-- 索引
CREATE INDEX idx_audit_session ON agent_audit_logs (session_id, event_time);
CREATE INDEX idx_audit_user ON agent_audit_logs (user_id, event_time);
CREATE INDEX idx_audit_task ON agent_audit_logs (task_id, event_time);
CREATE INDEX idx_audit_event_type ON agent_audit_logs (event_type, event_time);
CREATE INDEX idx_audit_risk ON agent_audit_logs (risk_level, event_time);

# audit-event-types.yaml
event_types:
  
  # 工具调用事件
  tool_call:
    description: "Agent 调用了某个工具"
    actions:
      - read_file          # 读取文件
      - write_file         # 写入文件
      - run_command        # 执行命令
      - search_code        # 搜索代码
      - create_pr          # 创建 PR
      - merge_pr           # 合并 PR
    payload_schema:
      tool_name: string
      parameters: object
      result_summary: string
      duration_ms: integer
      success: boolean
    
  # Prompt 事件
  prompt:
    description: "Agent 接收或生成了 Prompt"
    actions:
      - prompt_received    # 接收用户 Prompt
      - prompt_generated   # Agent 生成 Prompt
      - context_loaded     # 加载上下文包
    payload_schema:
      prompt_text: string          # 脱敏后的 Prompt
      token_count: integer
      context_files: list[string]
      has_sensitive_data: boolean
  
  # 代码变更事件
  diff:
    description: "Agent 产生了代码变更"
    actions:
      - file_created       # 创建文件
      - file_modified      # 修改文件
      - file_deleted       # 删除文件
    payload_schema:
      file_path: string
      change_type: string          # created / modified / deleted
      lines_added: integer
      lines_removed: integer
      diff_content: string         # 脱敏后的 diff
      affected_modules: list[string]
  
  # 审批事件
  approval:
    description: "操作需要或获得了审批"
    actions:
      - approval_requested  # 请求审批
      - approval_granted    # 审批通过
      - approval_denied     # 审批拒绝
      - approval_timeout    # 审批超时
    payload_schema:
      approval_id: string
      requester: string
      approver: string
      reason: string
      risk_level: string
      expires_at: timestamp
  
  # 成本事件
  cost:
    description: "API 调用产生的成本"
    actions:
      - api_call           # API 调用
    payload_schema:
      model: string
      input_tokens: integer
      output_tokens: integer
      cost_usd: decimal
      cache_hit: boolean

# app/audit/logger.py
"""
Agent 审计日志记录器。
记录所有 Agent 操作到数据库。
"""
import json
import hashlib
from datetime import datetime
from typing import Optional

class AuditLogger:
    def __init__(self, db_connection):
        self.db = db_connection
    
    def log_tool_call(self, session_id: str, task_id: str, user_id: str,
                      tool_name: str, parameters: dict, result_summary: str,
                      duration_ms: int, success: bool, risk_level: str = "LOW"):
        """记录工具调用"""
        
        # 脱敏处理
        safe_params = self._sanitize_parameters(tool_name, parameters)
        
        payload = {
            "tool_name": tool_name,
            "parameters": safe_params,
            "result_summary": result_summary[:500],  # 截断
            "duration_ms": duration_ms,
            "success": success,
        }
        
        self._insert_event(
            session_id=session_id,
            task_id=task_id,
            user_id=user_id,
            event_type="tool_call",
            event_action=tool_name,
            payload=payload,
            risk_level=risk_level,
        )
    
    def log_prompt(self, session_id: str, user_id: str, prompt_text: str,
                   token_count: int, context_files: list, is_user_prompt: bool):
        """记录 Prompt 事件"""
        
        # 脱敏：移除敏感信息
        sanitized_prompt = self._sanitize_prompt(prompt_text)
        
        payload = {
            "prompt_text": sanitized_prompt,
            "token_count": token_count,
            "context_files": context_files,
            "has_sensitive_data": sanitized_prompt != prompt_text,
            "is_user_prompt": is_user_prompt,
        }
        
        self._insert_event(
            session_id=session_id,
            user_id=user_id,
            event_type="prompt",
            event_action="prompt_received" if is_user_prompt else "prompt_generated",
            payload=payload,
            risk_level="LOW",
        )
    
    def log_diff(self, session_id: str, task_id: str, user_id: str,
                 file_path: str, change_type: str, diff_content: str,
                 lines_added: int, lines_removed: int):
        """记录代码变更"""
        
        # 脱敏：移除密钥、密码等
        sanitized_diff = self._sanitize_diff(diff_content)
        
        payload = {
            "file_path": file_path,
            "change_type": change_type,
            "lines_added": lines_added,
            "lines_removed": lines_removed,
            "diff_content": sanitized_diff,
        }
        
        # 根据文件路径判断风险等级
        risk = self._assess_diff_risk(file_path)
        
        self._insert_event(
            session_id=session_id,
            task_id=task_id,
            user_id=user_id,
            event_type="diff",
            event_action=f"file_{change_type}",
            payload=payload,
            risk_level=risk,
        )
    
    def log_approval(self, session_id: str, user_id: str, approval_id: str,
                     action: str, approver: str, reason: str, risk_level: str):
        """记录审批事件"""
        
        payload = {
            "approval_id": approval_id,
            "requester": user_id,
            "approver": approver,
            "reason": reason,
            "risk_level": risk_level,
        }
        
        self._insert_event(
            session_id=session_id,
            user_id=user_id,
            event_type="approval",
            event_action=action,
            payload=payload,
            risk_level=risk_level,
        )
    
    def log_cost(self, session_id: str, user_id: str, model: str,
                 input_tokens: int, output_tokens: int, cost_usd: float):
        """记录成本事件"""
        
        payload = {
            "model": model,
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
            "cost_usd": cost_usd,
        }
        
        self._insert_event(
            session_id=session_id,
            user_id=user_id,
            event_type="cost",
            event_action="api_call",
            payload=payload,
            cost_usd=cost_usd,
            tokens_used=input_tokens + output_tokens,
        )
    
    def _insert_event(self, session_id: str, task_id: Optional[str],
                      user_id: str, event_type: str, event_action: str,
                      payload: dict, risk_level: str,
                      cost_usd: float = None, tokens_used: int = None):
        """插入事件到数据库"""
        
        # 计算校验和（防篡改）
        payload_json = json.dumps(payload, sort_keys=True)
        checksum = hashlib.sha256(payload_json.encode()).hexdigest()
        
        self.db.execute("""
            INSERT INTO agent_audit_logs 
            (session_id, task_id, user_id, event_type, event_action, 
             payload, risk_level, cost_usd, tokens_used, checksum)
            VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
        """, (session_id, task_id, user_id, event_type, event_action,
              payload_json, risk_level, cost_usd, tokens_used, checksum))
    
    def _sanitize_parameters(self, tool_name: str, params: dict) -> dict:
        """脱敏工具参数"""
        safe = params.copy()
        
        # 移除敏感字段
        sensitive_keys = ["password", "secret", "token", "api_key", "private_key"]
        for key in sensitive_keys:
            if key in safe:
                safe[key] = "***REDACTED***"
        
        return safe
    
    def _sanitize_prompt(self, prompt: str) -> str:
        """脱敏 Prompt 内容"""
        import re
        # 移除 API Key 模式
        prompt = re.sub(r"sk-[a-zA-Z0-9]{32,}", "***API_KEY***", prompt)
        # 移除密码模式
        prompt = re.sub(r"password[:=]\s*\S+", "password=***", prompt)
        return prompt
    
    def _sanitize_diff(self, diff: str) -> str:
        """脱敏 diff 内容"""
        # 移除密钥、Token 等
        import re
        diff = re.sub(r"[\"'](?:sk-|pk-|ghp_)[a-zA-Z0-9]{32,}[\"']", '"***SECRET***"', diff)
        return diff
    
    def _assess_diff_risk(self, file_path: str) -> str:
        """根据文件路径评估风险等级"""
        high_risk_patterns = [
            "payment", "auth", "security", "crypto",
            "migrations", "schema", "config/prod",
        ]
        
        for pattern in high_risk_patterns:
            if pattern in file_path.lower():
                return "HIGH"
        
        medium_risk_patterns = ["controller", "service", "handler"]
        for pattern in medium_risk_patterns:
            if pattern in file_path.lower():
                return "MEDIUM"
        
        return "LOW"

-- 回放某个会话的完整操作序列
SELECT 
    event_time,
    event_type,
    event_action,
    payload->>'tool_name' as tool_name,
    payload->>'file_path' as file_path,
    risk_level,
    cost_usd
FROM agent_audit_logs
WHERE session_id = 'session-2024-06-15-001'
ORDER BY event_time;

-- 查询所有高风险操作
SELECT 
    event_time,
    user_name,
    event_action,
    payload->>'file_path' as file_path,
    payload->>'reason' as reason
FROM agent_audit_logs
WHERE risk_level IN ('HIGH', 'CRITICAL')
  AND event_time >= NOW() - INTERVAL '7 days'
ORDER BY event_time DESC;

-- 按用户统计成本
SELECT 
    user_name,
    COUNT(*) as call_count,
    SUM(cost_usd) as total_cost,
    SUM(tokens_used) as total_tokens,
    AVG(cost_usd) as avg_cost_per_call
FROM agent_audit_logs
WHERE event_type = 'cost'
  AND event_time >= NOW() - INTERVAL '30 days'
GROUP BY user_name
ORDER BY total_cost DESC;

# app/audit/compliance_export.py
"""
审计日志合规导出。
支持导出为 CSV、JSON、PDF 格式。
"""
import csv
import json
from datetime import date, timedelta

def export_audit_logs(start_date: date, end_date: date, 
                      format: str = "csv", output_path: str):
    """导出审计日志"""
    
    logs = db.query("""
        SELECT * FROM agent_audit_logs
        WHERE event_time BETWEEN %s AND %s
        ORDER BY event_time
    """, (start_date, end_date))
    
    if format == "csv":
        with open(output_path, 'w', newline='') as f:
            writer = csv.DictWriter(f, fieldnames=logs[0].keys())
            writer.writeheader()
            writer.writerows(logs)
    
    elif format == "json":
        with open(output_path, 'w') as f:
            json.dump(logs, f, indent=2, default=str)
    
    elif format == "pdf":
        # 生成 PDF 报告
        generate_pdf_report(logs, start_date, end_date, output_path)
    
    # 生成导出记录
    log_export_event(start_date, end_date, format, output_path)

# audit-retention-policy.yaml
retention:
  # 热存储（快速查询）
  hot_storage:
    duration_days: 90
    storage: "PostgreSQL"
    purpose: "日常查询、会话回放"
  
  # 温存储（归档查询）
  warm_storage:
    duration_days: 365
    storage: "S3 + Athena"
    purpose: "合规审查、事故复盘"
  
  # 冷存储（长期归档）
  cold_storage:
    duration_days: 2555  # 7 年
    storage: "S3 Glacier"
    purpose: "法律合规要求"
  
  # 清理策略
  cleanup:
    schedule: "daily"
    action: "move_to_next_tier"
    notification: true