docs(dev-workflow): 添加 Orchestrator 行为规范

参考 GSD (Get Shit Done) 模式，明确子代理完成时的强制行为：

关键改进：
1. Orchestrator + Subagent 架构说明
2. 子代理完成时的强制行为（伪代码）
3. 绝对禁止的行为清单
4. 必须的行为清单
5. 状态追踪文件格式

核心原则：汇报结果 → 判断阶段 → 立即启动下一步

dev-workflow/.stats_data.json

@@ -0,0 +1,21 @@
+{
+  "events": [
+    {
+      "timestamp": "2026-03-09T08:53:12.508206+00:00",
+      "event_type": "debate",
+      "mode": "cross",
+      "status": "success",
+      "metadata": {
+        "topic": "AI安全",
+        "password": "[REDACTED]"
+      }
+    },
+    {
+      "timestamp": "2026-03-09T08:53:12.540089+00:00",
+      "event_type": "review",
+      "mode": "normal",
+      "status": "failed",
+      "metadata": {}
+    }
+  ]
+}

dev-workflow/SKILL.md

@@ -0,0 +1,506 @@
+---
+name: dev-workflow
+description: "Spec-Driven 开发流程技能，支持 Cross-Debate 和 Cross-Review。当用户说'添加功能'、'开发XXX'时使用。"
+---
+
+# dev-workflow - Agent Engineering 开发流程
+
+## 什么时候使用
+
+当用户发送包含以下模式的消息时触发：
+
+| 触发词 | 行为 |
+|--------|------|
+| `添加功能` + 描述 | 开始 Agent Engineering 开发流程 |
+| `开发` + 描述 | 同上 |
+| `实现` + 描述 | 同上 |
+
+## 环境变量
+
+| 变量名 | 用途 | 必需 |
+|--------|------|------|
+| `GITEA_TOKEN` | Gitea API Token（创建 PR） | 是 |
+| `GITEA_URL` | Gitea 服务地址 | 否（默认 http://192.168.1.22:8418） |
+| `GITEA_REPO` | 仓库路径 | 否（默认 starmen1265/kb-skills） |
+
+---
+
+## 🎯 核心原则
+
+### 自动流程衔接
+
+**Phase 2-5 自动执行，不在中间阶段暂停等待用户！**
+
+```
+Phase 1 → 用户确认方案 → Phase 2 → Phase 3 → Phase 4 → Phase 5 → Phase 6 → 用户合并 PR
+         ↑                                                                    ↑
+      唯一介入点                                                          唯一介入点
+```
+
+### 用户介入点（只有 2 个）
+
+| 阶段 | 介入方式 | 说明 |
+|------|---------|------|
+| **Phase 1** | 方案选择 | Cross-Debate 后选择或调整方案 |
+| **Phase 6** | 最终审核 | 查看 PR，决定是否合并 |
+
+**其他阶段自动执行，不需要用户推动！**
+
+---
+
+## 工作流程（Cross 模式）
+
+```
+用户描述需求
+    ↓
+┌─────────────────────────────────────┐
+│  Phase 1: Cross-Debate（方案辩论）   │
+│  - 3个 Agent 分别出方案             │
+│  - 交叉审查对方方案                  │
+│  - 汇总统一方案                      │
+│  - ⏸️ 等待用户确认 ← 唯一介入点     │
+└─────────────────────────────────────┘
+    ↓ 用户确认后自动继续
+┌─────────────────────────────────────┐
+│  Phase 2: 分支创建                   │
+│  - 创建 feature/* 分支               │
+│  - 自动继续 →                        │
+└─────────────────────────────────────┘
+    ↓
+┌─────────────────────────────────────┐
+│  Phase 3: 子代理开发                 │
+│  - 基于 spec 启动开发子代理          │
+│  - ⚠️ 验证工作目录和代码            │
+│  - 本地测试通过                      │
+│  - 提交代码到功能分支                │
+│  - 自动继续 →                        │
+└─────────────────────────────────────┘
+    ↓
+┌─────────────────────────────────────┐
+│  Phase 4: Cross-Review（交叉审核）   │
+│  - 3个 Agent 交叉审核代码            │
+│  - 汇总问题列表                      │
+│  - 自动修复                          │
+│  - 多轮迭代直到通过                  │
+│  - 自动继续 →                        │
+└─────────────────────────────────────┘
+    ↓
+┌─────────────────────────────────────┐
+│  Phase 5: 安全审计                   │
+│  - 搜索安全最佳实践                  │
+│  - 审计权限/注入/密钥等问题          │
+│  - 修复安全漏洞                      │
+│  - 自动继续 →                        │
+└─────────────────────────────────────┘
+    ↓
+┌─────────────────────────────────────┐
+│  Phase 6: 创建 PR                    │
+│  - 调用 Gitea API 创建 PR            │
+│  - ⚠️ 禁止 auto-merge               │
+│  - ⏸️ 等待用户审核 ← 唯一介入点     │
+└─────────────────────────────────────┘
+    ↓
+用户决定是否合并 PR
+```
+
+---
+
+## 执行步骤详解
+
+### Phase 1: Cross-Debate（方案辩论）
+
+#### Step 1.1: 启动多 Agent 辩论
+
+使用 `sessions_spawn` 并行启动 3 个 Agent：
+
+```
+# Agent 1: Claude Code（偏保守，重视安全）
+sessions_spawn(
+  task="为以下需求设计方案。你是保守派，重视安全性和稳定性。
+        需求：{{用户需求}}
+        输出：1) 技术方案 2) 潜在风险 3) 实现步骤",
+  agentId="claude-code",
+  runtime="subagent"
+)
+
+# Agent 2: Codex（偏全面，考虑边界情况）
+sessions_spawn(
+  task="为以下需求设计方案。你是全面派，考虑所有边界情况。
+        需求：{{用户需求}}
+        输出：1) 技术方案 2) 边界情况 3) 实现步骤",
+  agentId="codex",
+  runtime="subagent"
+)
+
+# Agent 3: Gemini CLI（偏快速，先出方案再补细节）
+sessions_spawn(
+  task="为以下需求设计方案。你是快速派，优先快速实现。
+        需求：{{用户需求}}
+        输出：1) 技术方案 2) MVP范围 3) 实现步骤",
+  agentId="gemini",
+  runtime="subagent"
+)
+```
+
+#### Step 1.2: 交叉审查
+
+让三个 Agent 互相审查对方方案，承认盲点，补充建议。
+
+#### Step 1.3: 汇总统一方案
+
+收集审查结果后，生成**统一方案**（不是让用户选择 A/B/C）：
+
+```
+## 🗳️ Cross-Debate 结果
+
+### 共识点
+- ✅ 使用 XX 技术栈
+- ✅ 需要 YY 数据结构
+
+### 统一方案
+[经过辩论补充后的最终方案]
+
+✅ 确认开始实施？
+```
+
+#### Step 1.4: 用户确认
+
+**这是唯一的介入点！** 用户确认后，Phase 2-6 自动执行。
+
+---
+
+### Phase 2: 分支创建（自动执行）
+
+```bash
+git checkout main && git pull
+git checkout -b feature/<描述>
+```
+
+**完成后自动进入 Phase 3，不等待用户确认！**
+
+---
+
+### Phase 3: 子代理开发（自动执行）
+
+#### Step 3.1: 启动开发子代理
+
+```
+sessions_spawn(
+  task="""根据 spec 开发 {{功能名称}}
+
+  ## 工作目录（重要！）
+  在 {{REPO_PATH}} 目录下工作，不要使用其他目录！
+
+  ## Spec 文件
+  读取 {{REPO_PATH}}/.planning/spec.md
+
+  ## 要求
+  1. 只写目标，自己探索怎么实现
+  2. 本地测试通过后再提交
+  3. 提交信息遵循 Conventional Commits
+
+  ## 安全约束
+  - 不能 fail-open
+  - 并发请求不能超额
+  """,
+  cwd="{{REPO_PATH}}",  # 明确指定工作目录
+  runtime="subagent"
+)
+```
+
+#### Step 3.2: ⚠️ 验证子代理工作（关键步骤）
+
+**子代理可能：**
+1. 写入错误的工作目录
+2. 报告虚假结果
+
+**必须验证：**
+
+```python
+def verify_subagent_work(repo_path: str, expected_files: list) -> dict:
+    """验证子代理是否正确完成了工作"""
+    
+    # 1. 检查代码是否写入正确位置
+    for file in expected_files:
+        if not Path(f"{repo_path}/{file}").exists():
+            return {
+                "success": False,
+                "error": f"文件不存在于正确位置: {file}",
+                "hint": "子代理可能写入了错误的工作目录"
+            }
+    
+    # 2. 检查是否有新的 git 提交
+    result = run_cmd(f"cd {repo_path} && git log origin/main..HEAD --oneline")
+    if not result.stdout.strip():
+        return {
+            "success": False,
+            "error": "没有新的提交",
+            "hint": "子代理可能只报告了结果但没有实际开发"
+        }
+    
+    # 3. 运行测试
+    result = run_cmd(f"cd {repo_path} && python3 -m pytest")
+    if result.returncode != 0:
+        return {
+            "success": False,
+            "error": "测试未通过",
+            "hint": result.stderr
+        }
+    
+    return {"success": True}
+```
+
+**如果验证失败：**
+1. 检查子代理是否写入了 `~/.openclaw/agents/xxx/workspace/` 而不是目标仓库
+2. 手动修复或重新启动子代理
+
+**完成后自动进入 Phase 4，不等待用户确认！**
+
+---
+
+### Phase 4: Cross-Review（交叉审核）（自动执行）
+
+#### Step 4.1: 启动多 Agent 审核
+
+```
+sessions_spawn(task="审查代码变更...", runtime="subagent")
+sessions_spawn(task="审查代码变更...", runtime="subagent")
+sessions_spawn(task="审查测试覆盖...", runtime="subagent")
+```
+
+#### Step 4.2: 汇总问题并修复
+
+自动修复 Critical/High 问题，不需要用户确认。
+
+#### Step 4.3: 多轮迭代
+
+重复审核直到通过。
+
+**完成后自动进入 Phase 5，不等待用户确认！**
+
+---
+
+### Phase 5: 安全审计（自动执行）
+
+```
+sessions_spawn(
+  task="""安全审计代码...""",
+  runtime="subagent"
+)
+```
+
+自动修复安全问题。
+
+**完成后自动进入 Phase 6，不等待用户确认！**
+
+---
+
+### Phase 6: 创建 PR
+
+```bash
+python3 dev_workflow.py create-pr --branch feature/xxx --title "PR 标题"
+```
+
+**⚠️ 禁止 auto-merge**
+
+PR 创建后告知用户：
+
+```
+✅ PR #xxx 已创建
+
+**分支:** feature/xxx
+**链接:** http://192.168.1.22:8418/starmen1265/kb-skills/pulls/xxx
+
+**已完成：**
+- ✅ Cross-Debate（3个Agent辩论）
+- ✅ Cross-Review（3个Agent审核，X轮）
+- ✅ 安全审计
+
+**这是第二个介入点：** 请审核 PR，决定是否合并。
+```
+
+---
+
+## CLI 命令
+
+```bash
+# 创建 spec（从 Cross-Debate 结果）
+python3 dev_workflow.py create-spec --title "标题" --description "描述"
+
+# 从 spec 创建分支
+python3 dev_workflow.py create-branch --spec .planning/spec.md
+
+# 创建 PR（禁止 auto-merge）
+python3 dev_workflow.py create-pr --branch feature/xxx --title "PR标题"
+
+# 查看当前状态
+python3 dev_workflow.py status
+
+# 查看 Cross 模式统计
+python3 dev_workflow.py stats
+
+# 记录事件
+python3 dev_workflow.py record-event --type debate --mode cross
+```
+
+---
+
+## 🔄 Orchestrator 行为规范（关键！）
+
+**这是自动衔接的核心机制，参考 GSD (Get Shit Done) 模式。**
+
+### Orchestrator + Subagent 架构
+
+```
+┌─────────────────────────────────────┐
+│  Orchestrator (主会话)              │
+│  - 始终在主上下文中运行              │
+│  - 启动子代理                        │
+│  - 路由流程                          │
+│  - 永远不要停下来等待用户            │
+└─────────────────────────────────────┘
+           ↓ sessions_spawn
+┌─────────────────────────────────────┐
+│  Subagent (子代理)                  │
+│  - 新鲜上下文                        │
+│  - 专注任务                          │
+│  - 返回结构化结果                    │
+│  - 完成后通知 Orchestrator          │
+└─────────────────────────────────────┘
+           ↓ 完成事件
+┌─────────────────────────────────────┐
+│  Orchestrator 收到完成事件          │
+│  - 判断当前阶段                      │
+│  - 立即启动下一阶段                  │
+│  - 不要汇报后停下来！                │
+└─────────────────────────────────────┘
+```
+
+### 子代理完成时的强制行为
+
+当收到子代理完成事件时，**必须执行以下检查并立即行动**：
+
+```python
+# 伪代码 - Orchestrator 行为
+
+def on_subagent_complete(event):
+    # 1. 判断当前阶段
+    current_phase = get_current_phase()
+    
+    # 2. 根据阶段立即启动下一步
+    if current_phase == "Phase 3 开发":
+        # 验证代码
+        verify_code()
+        # 立即启动 Phase 4 - 不要等待用户！
+        spawn_review_agents()
+        
+    elif current_phase == "Phase 4 审查":
+        # 汇总问题
+        issues = aggregate_issues()
+        if issues:
+            # 立即启动修复子代理
+            spawn_fix_agent()
+        else:
+            # 立即启动 Phase 5 安全审计
+            spawn_security_audit()
+            
+    elif current_phase == "Phase 4 修复":
+        # 立即启动 Phase 5 安全审计
+        spawn_security_audit()
+        
+    elif current_phase == "Phase 5 安全审计":
+        issues = get_security_issues()
+        if issues:
+            # 立即启动安全修复
+            spawn_security_fix()
+        else:
+            # 立即推送分支
+            push_branch()
+            # 提供 PR 链接 - 这是第二个介入点
+            
+    elif current_phase == "Phase 5 安全修复":
+        # 立即推送分支
+        push_branch()
+        # 提供 PR 链接
+```
+
+### ⚠️ 绝对禁止的行为
+
+```
+❌ 汇报子代理结果后停下来等待
+❌ 说"完成了"然后等待用户说"继续"
+❌ 每个阶段都问用户"然后呢？"
+❌ 收到子代理完成事件后只转发结果
+```
+
+### ✅ 必须的行为
+
+```
+✅ 汇报结果 → 判断阶段 → 立即启动下一步
+✅ Phase 3 完成 → 验证 → Phase 4
+✅ Phase 4 完成 → 修复 → Phase 5
+✅ Phase 5 完成 → 推送 → PR 链接
+✅ 只有 Phase 1 和 Phase 6 才停下来
+```
+
+### 状态追踪（可选但推荐）
+
+在 `.planning/workflow-state.json` 中追踪状态：
+
+```json
+{
+  "current_phase": 4,
+  "phase_status": {
+    "1": "done",
+    "2": "done",
+    "3": "done",
+    "4": "in_progress",
+    "5": "pending",
+    "6": "pending"
+  },
+  "subagent_label": "review-1",
+  "started_at": "2026-03-09T16:00:00Z"
+}
+```
+
+---
+
+## ⚠️ 已知问题与改进
+
+### 问题 1: 子代理工作目录错误
+
+**现象**：子代理把代码写到 `~/.openclaw/agents/xxx/workspace/` 而不是目标仓库
+
+**解决方案**：
+1. 在任务描述中明确指定工作目录
+2. 在 Phase 3 增加验证步骤
+
+### 问题 2: 子代理虚假报告
+
+**现象**：子代理报告完成了开发，但实际没有提交任何代码
+
+**解决方案**：
+1. 在 Phase 3 增加验证步骤
+2. 检查 git log 确认有新提交
+
+---
+
+## 六条铁律
+
+1. **Briefing 写目标，不写文件** - 你是 PM，Agent 是工程师
+2. **Agent 共识需要人类验证** - 三个 AI 同意不代表正确
+3. **本地完成所有工作再推远端** - 实现、测试、审核、修复全在本地
+4. **安全约束要写进 Briefing** - "不能 fail-open"、"并发不能超额"
+5. **密钥永远不过聊天** - 用环境变量文件，命令里用变量名
+6. **Briefing 必须包含产品上下文** - 用户画像、产品定位、商业模式
+
+---
+
+## 注意事项
+
+1. 确保 `GITEA_TOKEN` 已配置
+2. 功能分支命名遵循规范
+3. 提交信息遵循 Conventional Commits
+4. 开发前先 pull 最新代码
+5. **禁止 auto-merge** - 流程错误比代码 bug 更贵
+6. **Phase 2-5 自动执行** - 不需要用户推动，只有 Phase 1 和 Phase 6 需要用户介入

dev-workflow/dev_workflow.py

@@ -0,0 +1,132 @@
+#!/usr/bin/env python3
+"""
+Dev Workflow CLI
+
+开发工作流命令行工具，包含 Cross 模式统计报告功能。
+"""
+
+import argparse
+import sys
+from pathlib import Path
+
+# 添加当前目录到路径
+sys.path.insert(0, str(Path(__file__).parent))
+
+from stats import record_event, get_stats, format_stats_report
+
+
+def cmd_stats(args):
+    """显示统计报告"""
+    stats = get_stats()
+    report = format_stats_report(stats)
+    print(report)
+
+
+def cmd_record_event(args):
+    """记录一个事件"""
+    # 构建 metadata
+    metadata = {}
+    if args.metadata:
+        for item in args.metadata:
+            if "=" in item:
+                key, value = item.split("=", 1)
+                metadata[key] = value
+    
+    success = record_event(
+        event_type=args.type,
+        mode=args.mode,
+        metadata=metadata if metadata else None,
+        status=args.status
+    )
+    
+    if success:
+        print(f"✅ 事件记录成功: type={args.type}, mode={args.mode}, status={args.status}")
+    else:
+        print(f"❌ 事件记录失败")
+        sys.exit(1)
+
+
+def cmd_clear_stats(args):
+    """清除统计数据（谨慎使用）"""
+    from stats import STATS_FILE
+    import os
+    
+    if STATS_FILE.exists():
+        if args.force or input("确认清除所有统计数据? [y/N]: ").lower() == "y":
+            os.remove(STATS_FILE)
+            print("✅ 统计数据已清除")
+        else:
+            print("操作已取消")
+    else:
+        print("统计数据文件不存在")
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="Dev Workflow - 开发工作流工具",
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+        epilog="""
+示例:
+  # 查看统计报告
+  python3 dev_workflow.py stats
+  
+  # 记录一个 Cross 模式辩论事件
+  python3 dev_workflow.py record-event --type debate --mode cross
+  
+  # 记录一个失败的事件（带元数据）
+  python3 dev_workflow.py record-event --type review --mode normal --status failed --metadata pr_id=123
+  
+  # 清除统计数据
+  python3 dev_workflow.py clear-stats --force
+"""
+    )
+    
+    subparsers = parser.add_subparsers(dest="command", help="可用命令")
+    
+    # stats 命令
+    stats_parser = subparsers.add_parser("stats", help="查看统计报告")
+    stats_parser.set_defaults(func=cmd_stats)
+    
+    # record-event 命令
+    record_parser = subparsers.add_parser("record-event", help="记录事件")
+    record_parser.add_argument(
+        "--type", "-t",
+        required=True,
+        choices=["debate", "review", "audit"],
+        help="事件类型: debate(辩论), review(评审), audit(审计)"
+    )
+    record_parser.add_argument(
+        "--mode", "-m",
+        required=True,
+        choices=["cross", "normal"],
+        help="运行模式: cross 或 normal"
+    )
+    record_parser.add_argument(
+        "--status", "-s",
+        default="success",
+        choices=["success", "failed"],
+        help="事件状态 (默认: success)"
+    )
+    record_parser.add_argument(
+        "--metadata",
+        nargs="*",
+        help="元数据，格式: key=value key2=value2"
+    )
+    record_parser.set_defaults(func=cmd_record_event)
+    
+    # clear-stats 命令
+    clear_parser = subparsers.add_parser("clear-stats", help="清除统计数据")
+    clear_parser.add_argument("--force", "-f", action="store_true", help="强制清除，不询问")
+    clear_parser.set_defaults(func=cmd_clear_stats)
+    
+    args = parser.parse_args()
+    
+    if args.command is None:
+        parser.print_help()
+        sys.exit(0)
+    
+    args.func(args)
+
+
+if __name__ == "__main__":
+    main()

dev-workflow/stats.py

@@ -0,0 +1,280 @@
+"""
+Cross 模式统计报告功能
+
+提供事件记录和统计报告生成功能，支持 Cross 和 Normal 两种模式。
+"""
+
+import json
+import os
+import fcntl
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Optional
+
+# 统计数据文件路径
+STATS_FILE = Path(__file__).parent / ".stats_data.json"
+
+# 敏感字段列表（用于过滤）
+SENSITIVE_FIELDS = {"password", "token", "secret", "api_key", "apikey", "credential"}
+
+
+def _filter_sensitive_data(data: dict) -> dict:
+    """过滤敏感数据
+    
+    递归过滤字典中的敏感字段，将值替换为 [REDACTED]
+    """
+    if not isinstance(data, dict):
+        return data
+    
+    filtered = {}
+    for key, value in data.items():
+        key_lower = key.lower().replace("-", "_").replace(" ", "_")
+        
+        # 检查是否为敏感字段
+        is_sensitive = any(s in key_lower for s in SENSITIVE_FIELDS)
+        
+        if is_sensitive:
+            filtered[key] = "[REDACTED]"
+        elif isinstance(value, dict):
+            filtered[key] = _filter_sensitive_data(value)
+        else:
+            filtered[key] = value
+    
+    return filtered
+
+
+def _ensure_stats_file():
+    """确保统计数据文件存在，并设置正确权限"""
+    if not STATS_FILE.exists():
+        STATS_FILE.write_text(json.dumps({"events": []}, ensure_ascii=False, indent=2))
+        os.chmod(STATS_FILE, 0o600)
+
+
+def _load_stats() -> dict:
+    """加载统计数据（带文件锁）"""
+    _ensure_stats_file()
+    
+    with open(STATS_FILE, "r") as f:
+        fcntl.flock(f.fileno(), fcntl.LOCK_SH)  # 共享锁
+        try:
+            data = json.load(f)
+        finally:
+            fcntl.flock(f.fileno(), fcntl.LOCK_UN)
+    
+    return data
+
+
+def _save_stats(data: dict):
+    """保存统计数据（带文件锁）"""
+    _ensure_stats_file()
+    
+    with open(STATS_FILE, "w") as f:
+        fcntl.flock(f.fileno(), fcntl.LOCK_EX)  # 排他锁
+        try:
+            json.dump(data, f, ensure_ascii=False, indent=2)
+        finally:
+            fcntl.flock(f.fileno(), fcntl.LOCK_UN)
+    
+    # 确保文件权限为 600
+    os.chmod(STATS_FILE, 0o600)
+
+
+def record_event(
+    event_type: str,
+    mode: str,
+    metadata: Optional[dict] = None,
+    status: str = "success"
+) -> bool:
+    """记录一个事件
+    
+    Args:
+        event_type: 事件类型 (debate, review, audit)
+        mode: 运行模式 (cross, normal)
+        metadata: 事件元数据（可选）
+        status: 事件状态 (success, failed)
+    
+    Returns:
+        bool: 记录是否成功
+    """
+    # 验证参数
+    valid_types = {"debate", "review", "audit"}
+    valid_modes = {"cross", "normal"}
+    valid_status = {"success", "failed"}
+    
+    if event_type not in valid_types:
+        print(f"错误: 无效的事件类型 '{event_type}'，有效值: {valid_types}")
+        return False
+    
+    if mode not in valid_modes:
+        print(f"错误: 无效的模式 '{mode}'，有效值: {valid_modes}")
+        return False
+    
+    if status not in valid_status:
+        print(f"错误: 无效的状态 '{status}'，有效值: {valid_status}")
+        return False
+    
+    # 过滤敏感数据
+    if metadata:
+        metadata = _filter_sensitive_data(metadata)
+    else:
+        metadata = {}
+    
+    # 创建事件记录
+    event = {
+        "timestamp": datetime.now(timezone.utc).isoformat(),
+        "event_type": event_type,
+        "mode": mode,
+        "status": status,
+        "metadata": metadata
+    }
+    
+    try:
+        data = _load_stats()
+        data["events"].append(event)
+        _save_stats(data)
+        return True
+    except Exception as e:
+        print(f"记录事件失败: {e}")
+        return False
+
+
+def get_stats() -> dict:
+    """获取统计汇总数据
+    
+    Returns:
+        dict: 统计汇总数据
+    """
+    data = _load_stats()
+    events = data.get("events", [])
+    
+    # 初始化统计结构
+    stats = {
+        "total_events": len(events),
+        "by_type": {
+            "debate": {"total": 0, "success": 0, "failed": 0},
+            "review": {"total": 0, "success": 0, "failed": 0},
+            "audit": {"total": 0, "success": 0, "failed": 0}
+        },
+        "by_mode": {
+            "cross": {"total": 0, "success": 0, "failed": 0},
+            "normal": {"total": 0, "success": 0, "failed": 0}
+        },
+        "cross_mode_breakdown": {
+            "debate": {"total": 0, "success": 0, "failed": 0},
+            "review": {"total": 0, "success": 0, "failed": 0},
+            "audit": {"total": 0, "success": 0, "failed": 0}
+        },
+        "normal_mode_breakdown": {
+            "debate": {"total": 0, "success": 0, "failed": 0},
+            "review": {"total": 0, "success": 0, "failed": 0},
+            "audit": {"total": 0, "success": 0, "failed": 0}
+        },
+        "success_rate": 0.0,
+        "first_event": None,
+        "last_event": None
+    }
+    
+    if not events:
+        return stats
+    
+    # 统计各类数据
+    for event in events:
+        event_type = event["event_type"]
+        mode = event["mode"]
+        status = event["status"]
+        
+        # 按类型统计
+        stats["by_type"][event_type]["total"] += 1
+        stats["by_type"][event_type][status] += 1
+        
+        # 按模式统计
+        stats["by_mode"][mode]["total"] += 1
+        stats["by_mode"][mode][status] += 1
+        
+        # 按模式+类型交叉统计
+        breakdown_key = f"{mode}_mode_breakdown"
+        stats[breakdown_key][event_type]["total"] += 1
+        stats[breakdown_key][event_type][status] += 1
+    
+    # 计算成功率
+    total = stats["total_events"]
+    success_count = sum(t["success"] for t in stats["by_type"].values())
+    stats["success_rate"] = round(success_count / total * 100, 2) if total > 0 else 0.0
+    
+    # 时间范围
+    stats["first_event"] = events[0]["timestamp"]
+    stats["last_event"] = events[-1]["timestamp"]
+    
+    return stats
+
+
+def format_stats_report(stats: dict) -> str:
+    """格式化统计报告为可读字符串
+    
+    Args:
+        stats: get_stats() 返回的统计数据
+    
+    Returns:
+        str: 格式化的统计报告
+    """
+    lines = []
+    lines.append("=" * 50)
+    lines.append("       Cross 模式统计报告")
+    lines.append("=" * 50)
+    
+    # 总览
+    lines.append(f"\n📊 总览")
+    lines.append(f"  总事件数: {stats['total_events']}")
+    lines.append(f"  成功率: {stats['success_rate']}%")
+    if stats['first_event']:
+        lines.append(f"  时间范围: {stats['first_event']} ~ {stats['last_event']}")
+    
+    # 按模式统计
+    lines.append(f"\n📈 按模式统计")
+    for mode, data in stats["by_mode"].items():
+        mode_name = "Cross 模式" if mode == "cross" else "Normal 模式"
+        rate = round(data["success"] / data["total"] * 100, 2) if data["total"] > 0 else 0
+        lines.append(f"  {mode_name}:")
+        lines.append(f"    总数: {data['total']} | 成功: {data['success']} | 失败: {data['failed']} | 成功率: {rate}%")
+    
+    # 按事件类型统计
+    lines.append(f"\n📋 按事件类型统计")
+    for event_type, data in stats["by_type"].items():
+        type_name = {"debate": "辩论", "review": "评审", "audit": "审计"}.get(event_type, event_type)
+        rate = round(data["success"] / data["total"] * 100, 2) if data["total"] > 0 else 0
+        lines.append(f"  {type_name}:")
+        lines.append(f"    总数: {data['total']} | 成功: {data['success']} | 失败: {data['failed']} | 成功率: {rate}%")
+    
+    # Cross 模式详细分解
+    lines.append(f"\n🔍 Cross 模式详细分解")
+    for event_type, data in stats["cross_mode_breakdown"].items():
+        type_name = {"debate": "辩论", "review": "评审", "audit": "审计"}.get(event_type, event_type)
+        rate = round(data["success"] / data["total"] * 100, 2) if data["total"] > 0 else 0
+        lines.append(f"  {type_name}:")
+        lines.append(f"    总数: {data['total']} | 成功: {data['success']} | 失败: {data['failed']} | 成功率: {rate}%")
+    
+    # Normal 模式详细分解
+    lines.append(f"\n🔍 Normal 模式详细分解")
+    for event_type, data in stats["normal_mode_breakdown"].items():
+        type_name = {"debate": "辩论", "review": "评审", "audit": "审计"}.get(event_type, event_type)
+        rate = round(data["success"] / data["total"] * 100, 2) if data["total"] > 0 else 0
+        lines.append(f"  {type_name}:")
+        lines.append(f"    总数: {data['total']} | 成功: {data['success']} | 失败: {data['failed']} | 成功率: {rate}%")
+    
+    lines.append("\n" + "=" * 50)
+    
+    return "\n".join(lines)
+
+
+if __name__ == "__main__":
+    # 测试代码
+    print("测试 stats.py 模块...")
+    
+    # 记录一些测试事件
+    record_event("debate", "cross", {"topic": "AI 安全"}, "success")
+    record_event("review", "normal", {"pr_id": 123}, "success")
+    record_event("audit", "cross", {"password": "secret123"}, "failed")  # 敏感数据会被过滤
+    
+    # 获取并打印统计
+    stats = get_stats()
+    print(format_stats_report(stats))