Edict 三省六部制多 Agent 任务分发流转体系：状态机、权限矩阵与调度架构全解

本文基于 Edict（三省六部制 · OpenClaw Multi-Agent Orchestration System）的 任务分发流转架构文档 展开，系统阐述一个制度化 AI 多 Agent 框架如何从业务制度设计（分权制衡的帝国官僚体系）落到代码实现（状态机、权限矩阵、自动派发、停滞恢复调度），并打通 React 看板、实时 API 与 OpenClaw 运行时。读完本文，你将掌握：三省六部任务从"皇上"下旨到六部并行执行、门下封驳、尚书汇总、最终回奏的完整流转链路；任务规格书与业务契约的字段语义；kanban_update.py 的 7+ 核心命令与权限约束；以及超时重试、停滞升级、自动回滚三层调度恢复机制，并可对照仓库源码（edict/backend/app/models/task.py、scripts/kanban_update.py 等）直接落地实践。

---

第一部分：业务架构 — 分权制衡的制度化协作

1.1 从"自由协作"到"制度化协作"的设计哲学

传统的多 Agent 框架（如 CrewAI、AutoGen）通常采用自由协作模式：Agent 自主选择协作对象、框架仅提供通信通道、质量控制完全依赖 Agent 智能。这种模式的典型问题包括：Agent 之间相互制造假数据、重复劳动、方案质量无保障。

三省六部体系反其道而行，采用制度化协作模式，模仿古代帝国官僚体系，通过"分权制衡 + 制度约束"把业务流程固化为不可绕过的制度：

              皇上 (User)
               │ ↓
             太子 (Taizi) —— [分拣官、消息接入总负责]
      ├─ 识别：这是旨意还是闲聊？
      ├─ 执行：直接回复闲聊 || 建立任务→转中书
      └─ 权限：只能调用 中书省
               │ ↓
           中书省 (Zhongshu) —— [规划官、方案起草总负责]
      ├─ 接旨后分析需求，拆解为子任务（todos）
      ├─ 调用门下省审议 OR 尚书省咨询
      └─ 权限：只能调用 门下 + 尚书
               │ ↓
           门下省 (Menxia) —— [审议官、质量把握人]
      ├─ 审查中书方案（可行性、完整性、风险）
      ├─ 准奏 OR 封驳（含修改建议）；若封驳 → 返回中书修改 → 重新审议（最多3轮）
      └─ 权限：只能调用 尚书 + 回调中书
               │ ↓ (✅ 准奏)
           尚书省 (Shangshu) —— [派发官、执行总指挥]
      ├─ 分析派发给哪个部门，调用六部执行，监控各部进度 → 汇总结果
      └─ 权限：只能调用 六部（不能越权调中书）
               │ ↓
               ├─ 礼部 (Libu)      - 文档编制官
               ├─ 户部 (Hubu)      - 数据分析官
               ├─ 兵部 (Bingbu)    - 代码实现官
               ├─ 刑部 (Xingbu)    - 测试审查官
               ├─ 工部 (Gongbu)    - 基础设施官
               └─ 吏部 (Libu_hr)   - 人力资源官
               │ ↓ (各部并行执行)
           尚书省·汇总 → 状态转 Review → 回调中书省转报皇上
           中书省·回奏 → 状态转 Done → 回复飞书消息给皇上

这套制度提供 4 大保障：

保障机制	实现细节	防护效果
制度性审核	门下省必审议所有中书方案，不可跳过	防止 Agent 胡乱执行，确保方案具有可行性
分权制衡	权限矩阵：谁能调谁严格定义	防止权力滥用（如尚书越权调中书改方案）
完全可观测	任务看板 10 个面板 + 每条任务数十条活动记录	实时看到任务卡在哪、谁在工作、工作状态如何
实时可干预	看板内一键 stop/cancel/resume/advance	紧急情况（如发现 Agent 走错方向）能立即纠正

1.2 任务完整流转流程

状态流转图（Mermaid stateDiagram）：

stateDiagram-v2
[*] --> Pending: 皇上下旨
Pending --> Taizi: 太子接旨
Taizi --> Zhongshu: 太子转交中书
Zhongshu --> Menxia: 中书提交审议
Menxia --> Zhongshu: 门下封驳(可多次)
Menxia --> Assigned: 门下准奏
Assigned --> Doing: 尚书派发执行
Doing --> Review: 各部完成
Review --> Done: 皇上御批通过
Review --> Menxia: 皇上要求修改
Done --> [*]
Doing --> [*]: 手动取消
Review --> [*]: 业务终止

✅ 理想路径（无阻滞，约 4-5 天完成）：

DAY 1:
  10:00 - 皇上飞书："为三省六部编写完整自动化测试方案"
          太子接旨。state = Taizi, org = 太子；自动派发 taizi agent
  10:30 - 太子分拣完毕，判定为「工作旨意」（非闲聊）
          建任务 JJC-20260228-E2E；flow_log 记录 "皇上 → 太子：下旨"
          state: Taizi → Zhongshu, org: 太子 → 中书省；自动派发 zhongshu agent

DAY 2:
  09:00 - 中书省接旨，开始规划，汇报进展（progress_log 记录）
  15:00 - 中书省完成方案；todos 快照（需求分析✅、方案设计✅、待审议🔄）
          flow_log 记录 "中书省 → 门下省：方案提交审议"
          state: Zhongshu → Menxia；自动派发 menxia agent

DAY 3:
  09:00 - 门下省开始审议（审查完整性和风险）
  14:00 - 判定 "方案可行，但缺失 _infer_agent_id_from_runtime 函数的测试"
          ✅ 准奏（带修改建议）；state: Menxia → Assigned；自动派发 shangshu agent

DAY 4:
  10:00 - 尚书省分析 "该测试方案应派给工部+刑部+礼部协力完成"
          state: Assigned → Doing；自动派发 bingbu/xingbu/libu 三个 agent（并行）

DAY 4-5:
  （各部并行执行 + hourly progress 实时汇报）
  - 兵部："✅ 已实现 16 个单元测试"
  - 刑部："🔄 正在编写集成测试（8/12 完成）"
  - 礼部："等待兵部完成再写报告"

DAY 5:
  14:00 - 各部完成；state: Doing → Review；尚书省汇总 "通过率 98.5%"
  15:00 - 中书省回奏；state: Review → Done；模板回复飞书，含成果链接和总结

❌ 挫折路径（含封驳和重试，约 6-7 天）：

DAY 3 [封驳场景]:
  14:00 - 门下省判定 "方案不完整，缺少性能测试 + 压力测试"，🚫 封驳
          review_round += 1；state: Menxia → Zhongshu（返回中书修改）
          自动派发 zhongshu agent（重新规划）
DAY 3-4:
  16:00 - 中书省收到封驳通知（唤醒 agent），补充性能测试方案
          state: Zhongshu → Menxia；自动派发 menxia agent（第2轮审议）
  18:00 - 门下省重新审议 ✅ 通过；state: Menxia → Assigned → Doing
DAY 7：全部完成（比理想路径晚 1-2 天）

1.3 任务规格书（Task Schema）与业务契约

任务是流转体系的核心数据结构，以下字段构成任务规格书（JSON 示意）：

{
  "id": "JJC-20260228-E2E",          // 任务全局唯一ID (JJC-日期-序号)
  "title": "为三省六部编写完整自动化测试方案",
  "official": "中书令",              // 负责官职
  "org": "中书省",                   // 当前负责部门
  "state": "Assigned",               // 当前状态（见 _STATE_FLOW）

  // ──── 质量与约束 ────
  "priority": "normal",              // 优先级：critical/high/normal/low
  "block": "无",                     // 当前阻滞原因（如"等待工部反馈"）
  "reviewRound": 2,                  // 门下审议第几轮
  "_prev_state": "Menxia",           // 若被 stop，记录之前状态用于 resume

  // ──── 业务产出 ────
  "output": "",                      // 最终任务成果（URL/文件路径/总结）
  "ac": "",                          // Acceptance Criteria（验收标准）

  // ──── 流转记录 ────
  "flow_log": [
    {"at": "2026-02-28T10:00:00Z", "from": "皇上", "to": "太子", "remark": "下旨：为三省六部编写完整自动化测试方案"},
    {"at": "2026-03-01T09:00:00Z", "from": "门下省", "to": "中书省", "remark": "🚫 封驳：需补充性能测试"},
    {"at": "2026-03-01T20:00:00Z", "from": "门下省", "to": "尚书省", "remark": "✅ 准奏通过（第2轮，5条建议已采纳）"}
  ],

  // ──── Agent 实时汇报 ────
  "progress_log": [
    {
      "at": "2026-02-28T10:35:00Z",
      "agent": "zhongshu", "agentLabel": "中书省",
      "text": "已接旨。分析测试需求，拟定三层测试方案...",
      "state": "Zhongshu", "org": "中书省",
      "tokens": 4500, "cost": 0.0045, "elapsed": 120,
      "todos": [
        {"id": "1", "title": "需求分析", "status": "completed"},
        {"id": "2", "title": "方案设计", "status": "in-progress"},
        {"id": "3", "title": "await审议", "status": "not-started"}
      ]
    }
  ],

  // ──── 调度元数据 ────
  "_scheduler": {
    "enabled": true,
    "stallThresholdSec": 180,         // 停滞超过180秒自动升级
    "maxRetry": 1,                    // 自动重试最多1次
    "retryCount": 0,
    "escalationLevel": 0,             // 0=无升级 1=门下协调 2=尚书协调
    "lastProgressAt": "2026-03-01T20:00:00Z",
    "stallSince": null,
    "lastDispatchStatus": "success",  // queued|success|failed|timeout|error
    "snapshot": {"state": "Assigned", "org": "尚书省", "note": "review-before-approve"}
  },

  // ──── 生命周期 ────
  "archived": false,
  "now": "门下省准奏，移交尚书省派发",
  "updatedAt": "2026-03-01T20:00:00Z"
}

在 Postgres 版后端中，这些字段被建模为 Task 表的列（JSONB 类型），见 edict/backend/app/models/task.py：flow_log、progress_log、todos、scheduler 均为 JSONB，并额外提供 trace_id（追踪链路 ID）、template_id、ac、target_dept、meta 等扩展字段；同时保留了 org、official、now、block、output、archived 等"旧看板兼容字段"，避免新后端与现有前端/迁移数据脱节。

业务契约：

契约	含义	违反后果
不可越级	太子只能调中书，中书只能调门下/尚书，六部不能对外调用	超权调用被拒绝，系统自动拦截
状态单向递进	Pending → Taizi → Zhongshu → ... → Done，不能跳过或倒退	只能通过 review_action(reject) 返回上一步
门下必审	所有中书提出的方案都要门下省审议，无法跳过	中书不能直接转尚书，门下必入
一旦Done无改	任务进入 Done/Cancelled 后不能再修改状态	若需修改需要创建新任务或取消后重新建
task_id唯一性	JJC-日期-序号 全局唯一，同一天同一任务不重复建	看板防重，自动去重
资源消耗透明	每次进展汇报都要上报 tokens/cost/elapsed	便于成本核算和性能优化

---

第二部分：技术架构 — 状态机、权限矩阵、数据融合与调度

2.1 状态机与自动派发

状态转移的权威定义（Single Source of Truth）

后端模型将状态机定义为枚举 + 转移表。状态枚举见 edict/backend/app/models/task.py：Taizi / Zhongshu / Menxia / Assigned / Next / Doing / Review / Done / Blocked / Cancelled / Pending / PendingConfirm，其中 TERMINAL_STATES = {Done, Cancelled} 为终态。

转移表 STATE_TRANSITIONS（task.py）比文档中的主链更完整，实际还包含 Next、Blocked、PendingConfirm 三个扩展状态：

STATE_TRANSITIONS = {
    TaskState.Pending: {TaskState.Taizi, TaskState.Cancelled},
    TaskState.Taizi: {TaskState.Zhongshu, TaskState.Cancelled},
    TaskState.Zhongshu: {TaskState.Menxia, TaskState.Cancelled, TaskState.Blocked},
    TaskState.Menxia: {TaskState.Assigned, TaskState.Zhongshu, TaskState.Cancelled},
    TaskState.Assigned: {TaskState.Doing, TaskState.Next, TaskState.Cancelled, TaskState.Blocked},
    TaskState.Next: {TaskState.Doing, TaskState.Cancelled, TaskState.Blocked},
    TaskState.Doing: {TaskState.Review, TaskState.Done, TaskState.Blocked, TaskState.Cancelled},
    TaskState.Review: {TaskState.Done, TaskState.Menxia, TaskState.Doing, TaskState.Cancelled, TaskState.PendingConfirm},
    TaskState.PendingConfirm: {TaskState.Done, TaskState.Review, TaskState.Cancelled},
    TaskState.Blocked: { /* 可恢复到任意工作态 */ },
}

注意几个关键差异：

• Menxia 可以回退到 Zhongshu（封驳路径）；
• Review 可以回退到 Menxia（皇上要求修改）或 Doing（返回执行），也可进入 PendingConfirm（高风险操作确认中间态）；
• Blocked 是一个可恢复的停滞态，可恢复到 Taizi/Zhongshu/Menxia/Assigned/Next/Doing/Review 任一工作态。

每个状态自动关联 Agent ID（STATE_AGENT_MAP，见 task.py）：

STATE_AGENT_MAP = {
    TaskState.Taizi: "taizi",
    TaskState.Zhongshu: "zhongshu",
    TaskState.Menxia: "menxia",
    TaskState.Assigned: "shangshu",
    TaskState.Review: "shangshu",
    TaskState.PendingConfirm: "shangshu",
    TaskState.Pending: "zhongshu",
}

Doing/Next 状态没有固定 Agent，而是通过 ORG_AGENT_MAP 从 task.org（部门）推断具体六部 agent：户部→hubu、礼部→libu、兵部→bingbu、刑部→xingbu、工部→gongbu、吏部→libu_hr（task.py）。STATE_ORG_MAP 则负责状态→部门的反向映射（如 Menxia → 门下省）。

关键实现事实：文档中 kanban_update.py state 命令的状态合法性校验并非硬编码，而是运行时从 task.py 动态解析权威转移表——_load_canonical_transitions() 通过正则提取 STATE_TRANSITIONS 代码块后 exec 求值（见 scripts/kanban_update.py），确保 JSON 看板模式与 Postgres 后端模式的转移规则永远一致（Single Source of Truth）。当 edict/backend 目录不存在时才使用内置 fallback 表。

自动派发流程

当任务状态转移时（通过 handle_advance_state() 或审批），后台自动执行派发：

1. 状态转移触发派发
   ├─ 查表 _STATE_AGENT_MAP 得到目标 agent_id
   ├─ 若是 Doing/Next，从 task.org 查表 _ORG_AGENT_MAP 推断具体部门 agent
   └─ 若无法推断则跳过派发（如 Done/Cancelled）

2. 构造派发消息（针对性促使 Agent 立即工作）
   ├─ taizi: "📜 皇上旨意需要你处理..."
   ├─ zhongshu: "📜 旨意已到中书省，请起草方案..."
   ├─ menxia: "📋 中书省方案提交审议..."
   ├─ shangshu: "📮 门下省已准奏，请派发执行..."
   └─ 六部: "📌 请处理任务..."

3. 后台异步派发（非阻塞）
   ├─ spawn daemon thread；标记 _scheduler.lastDispatchStatus = 'queued'
   ├─ 检查 Gateway 进程是否开启
   ├─ 运行 openclaw agent --agent {id} -m "{msg}" --deliver --timeout 300
   ├─ 重试最多2次（失败间隔5秒退避）
   └─ 更新 _scheduler 状态和错误信息

4. 派发状态转移
   ├─ success: lastDispatchStatus = 'success'
   ├─ failed: 记录失败原因（不 block 看板）
   ├─ timeout: 标记 timeout，允许用户手动重试/升级
   ├─ gateway-offline: 跳过此次派发（后续可重试）
   └─ error: 记录堆栈供调试

5. 到达目标 Agent 的处理
   ├─ Agent 从飞书消息收到通知
   ├─ 通过 kanban_update.py 与看板交互（更新状态/记录进展）
   └─ 完成工作后再次触发派发到下一个 Agent

新版后端的事件驱动实现（对应旧架构的 daemon 线程 + subprocess）在 edict/backend/app/workers/orchestrator_worker.py 与 edict/backend/app/workers/dispatch_worker.py 中：

• OrchestratorWorker 监听 task.created / task.status / task.completed / task.stalled 四个 topic：任务创建 → 派发太子；状态变更 → 查 STATE_AGENT_MAP 自动派发下游 agent；进入 Assigned 且带 assignee_org 时直接派发对应六部 agent，否则派发给 shangshu 让其人工路由（orchestrator_worker.py）。
• DispatchWorker 消费 task.dispatch stream，真正调用 openclaw agent --agent {id} -m "{msg}"（超时 300 秒），并做了快/慢 Agent 分桶并发控制：fast 桶（taizi/zhongshu/menxia/shangshu/zaochao，并发上限 4）与 slow 桶（六部，并发上限 3）互不阻塞（dispatch_worker.py）。
• 可靠性升级：旧架构用 daemon 线程 + subprocess.run，进程被杀后派发丢失；新架构基于 Redis Streams ACK 机制，worker 崩溃后未 ACK 的事件会被其他消费者自动认领（claim_stale + _recover_pending），"永不丢失"（见 dispatch_worker.py 与文件头部注释）。

富上下文注入：派发时 _build_task_context 将任务 ID/标题/状态/部门/优先级/todos/最近流转/最近进展/阻塞信息拼进消息；_build_reminder 按状态注入动态提醒（如 Menxia 状态提醒"通过则流转 Assigned，不通过则退回 Zhongshu"）；_build_memory_context 注入三级记忆（全局规则 → Agent 永久记忆 → 任务上下文决策链）；_load_agent_skills 按任务 tags/org 延迟加载 Skills（dispatch_worker.py）。上下文还通过环境变量（EDICT_TASK_ID、EDICT_TRACE_ID、EDICT_API_URL 等）和临时上下文文件（EDICT_CONTEXT_FILE）传递给 OpenClaw CLI（dispatch_worker.py）。

并发安全：状态流转在 transition_state() 中使用 SELECT ... FOR UPDATE 行级排他锁串行化同一任务的并发写入，防止并发 flow_log 丢失；任务创建/流转事件与任务本体写入同一事务的 outbox 表，由 OutboxRelay 异步投递，保证 DB 与 Event 原子一致（edict/backend/app/services/task_service.py）。

2.2 权限矩阵与 Subagent 调用

权限定义（agents.json / openclaw.json 中配置）

仓库根目录的 agents.json 是真实生效的权限矩阵配置，每个 agent 的 subagents.allowAgents 精确限定其可调用的下级 Agent：

[
  { "id": "taizi",   "subagents": { "allowAgents": ["zhongshu"] } },
  { "id": "zhongshu","subagents": { "allowAgents": ["menxia", "shangshu"] } },
  { "id": "menxia",  "subagents": { "allowAgents": ["shangshu", "zhongshu"] } },
  { "id": "shangshu","subagents": { "allowAgents": ["zhongshu", "menxia", "hubu", "libu", "bingbu", "xingbu", "gongbu", "libu_hr"] } },
  { "id": "hubu",    "subagents": { "allowAgents": ["shangshu"] } },
  { "id": "libu",    "subagents": { "allowAgents": ["shangshu"] } },
  { "id": "bingbu",  "subagents": { "allowAgents": ["shangshu"] } },
  { "id": "xingbu",  "subagents": { "allowAgents": ["shangshu"] } },
  { "id": "gongbu",  "subagents": { "allowAgents": ["shangshu"] } },
  { "id": "libu_hr", "subagents": { "allowAgents": ["shangshu"] } },
  { "id": "zaochao", "subagents": { "allowAgents": [] } }
]

权限检查机制（代码层面）

在 dispatch_for_state() 之外，还有一套防御性的权限检查：

def can_dispatch_to(from_agent, to_agent):
    """检查 from_agent 是否有权调用 to_agent。"""
    cfg = read_json(DATA / 'agent_config.json', {})
    agents = cfg.get('agents', [])
    from_record = next((a for a in agents if a.get('id') == from_agent), None)
    if not from_record:
        return False, f'{from_agent} 不存在'
    allowed = from_record.get('allowAgents', [])
    if to_agent not in allowed:
        return False, f'{from_agent} 无权调用 {to_agent}（允许列表：{allowed}）'
    return True, 'OK'

CLI 层的命令级权限策略（更细粒度的第二道防线）在 scripts/kanban_update.py 中以 AGENT_POLICY 表定义：协调类 Agent（taizi/zhongshu/menxia/shangshu）可执行 state/flow/progress/todo/memory/task-memo 等命令，其中 taizi 还可 create、shangshu 还可 delegate/shared-memo/confirm；执行类 Agent（六部）只能执行 progress/todo/done/block/memory/task-memo/delegate-result，不能 state/flow——即六部不能自己改状态或写流转记录，只能汇报进展和提交完成，从工具入口杜绝越权。越权调用会写入 audit_log.json 并直接 sys.exit(1) 拒绝。

权限违反示例与处理

场景	请求	结果	理由
正常	中书省 → 门下省审议	✅ 允许	门下在中书的 allowAgents 中
违反	中书省 → 尚书省改方案	❌ 拒绝	中书只能调门下/尚书，不能手工改尚书工作
正常	工部 → 尚书省 "我完成了"	✅ 改状态	通过 flow_log 和 progress_log（不是跨 Agent 调用）
违反	尚书省 → 中书省 "重新改方案"	❌ 拒绝	尚书不在门下/中书的 allowAgents 中
防控	Agent 伪造其他 agent 派发	❌ 拦截	API 层验证 HTTP 请求来源/签名

此外，kanban_update.py 的 _infer_agent_id_from_runtime() 会依次从环境变量（OPENCLAW_AGENT_ID 等）、工作目录名（workspace-*）、脚本路径推断当前执行 Agent 的身份（kanban_update.py），作为权限校验的身份依据。

2.3 数据融合：flow_log + progress_log + session JSONL

三层数据源

1️⃣ flow_log
   └─ 纯粹记录状态转移（Zhongshu → Menxia）
   └─ 数据源：任务 JSON 的 flow_log 字段；来自 Agent 通过 kanban_update.py flow 命令上报

2️⃣ progress_log
   └─ Agent 的实时工作汇报（文本进展、todos快照、资源消耗）
   └─ 数据源：任务 JSON 的 progress_log 字段；来自 kanban_update.py progress 命令上报
   └─ 周期：通常每30分钟或关键节点上报1次

3️⃣ session JSONL
   └─ Agent 的内部思考过程（thinking）、工具调用（tool_result）、对话历史（user）
   └─ 数据源：~/.openclaw/agents/{agent_id}/sessions/*.jsonl
   └─ 来自：OpenClaw 框架自动记录，Agent 无需主动操作
   └─ 周期：消息级别，粒度最细

问题诊断与解决方案

过去只靠 flow_log + progress_log 展现进展，存在"黑盒"问题：看不到 Agent 具体思考过程、工具调用结果、中间对话历史（例如 progress_log 记录"正在分析需求"，用户却看不到分析了什么）。

解决方案是在 get_task_activity() 中新增 session JSONL 融合逻辑（约 40 行）：

def get_task_activity(task_id):
    # ... 前面代码同上 ...

    # ── 融合 Agent Session 活动（thinking / tool_result / user）──
    session_entries = []

    # 活跃任务：尝试按 task_id 精确匹配
    if state not in ('Done', 'Cancelled'):
        if agent_id:
            entries = get_agent_activity(agent_id, limit=30, task_id=task_id)
            session_entries.extend(entries)
        # 也从相关 Agent 获取
        for ra in related_agents:
            if ra != agent_id:
                entries = get_agent_activity(ra, limit=20, task_id=task_id)
                session_entries.extend(entries)
    else:
        # 已完成任务：基于关键词匹配
        title = task.get('title', '')
        keywords = _extract_keywords(title)
        if keywords:
            for ra in related_agents[:5]:
                entries = get_agent_activity_by_keywords(ra, keywords, limit=15)
                session_entries.extend(entries)

    # 去重（通过 at+kind 去重避免重复）
    existing_keys = {(a.get('at', ''), a.get('kind', '')) for a in activity}
    for se in session_entries:
        key = (se.get('at', ''), se.get('kind', ''))
        if key not in existing_keys:
            activity.append(se)
            existing_keys.add(key)

    # 重新排序
    activity.sort(key=lambda x: x.get('at', ''))

    # 返回时标记数据来源
    return {
        'activity': activity,
        'activitySource': 'progress+session',  # 新标记
        # ... 其他字段 ...
    }

Session JSONL 格式解析

从 JSONL 中提取的条目统一转换为看板活动条目：

def _parse_activity_entry(item):
    """将 session jsonl 的 message 统一解析成看板活动条目。"""
    msg = item.get('message', {})
    role = str(msg.get('role', '')).strip().lower()
    ts = item.get('timestamp', '')

    # 🧠 Assistant 角色 - Agent思考过程
    if role == 'assistant':
        entry = {
            'at': ts,
            'kind': 'assistant',
            'text': '...主回复...',
            'thinking': '💭 Agent考虑到...',  # 内部思维链
            'tools': [
                {'name': 'bash', 'input_preview': 'cd /src && npm test'},
                {'name': 'file_read', 'input_preview': 'dashboard/server.py'},
            ]
        }
        return entry

    # 🔧 Tool Result - 工具调用结果
    if role in ('toolresult', 'tool_result'):
        entry = {
            'at': ts,
            'kind': 'tool_result',
            'tool': 'bash',
            'exitCode': 0,
            'output': '✓ All tests passed (123 tests)',
            'durationMs': 4500  # 执行时长
        }
        return entry

    # 👤 User - 人工反馈或对话
    if role == 'user':
        entry = {
            'at': ts,
            'kind': 'user',
            'text': '请实现测试用例的异常处理'
        }
        return entry

融合后的活动流结构

单个任务的 59 条活动流（JJC-20260228-E2E 示例）：

kind    count  代表事件
────────────────────────────────────────────────
flow      10   状态转移链（Pending→Taizi→Zhongshu→...）
progress  11   Agent工作汇报（"正在分析"、"已完成"）
todos     11   待办任务快照（进度更新时每条）
user       1   用户反馈（如"需要补充性能测试"）
assistant 10   Agent思考过程（💭 reasoning chain）
tool_result 16   工具调用记录（bash运行结果、API调用结果）
────────────────────────────────────────────────
总计      59   完整工作轨迹

看板展示时，用户可以：

• 📋 看流转链了解任务在哪个阶段
• 📝 看 progress 了解 Agent 实时说了什么
• ✅ 看 todos 了解任务拆解和完成进度
• 💭 看 assistant/thinking 了解 Agent 的思考过程
• 🔧 看 tool_result 了解每次工具调用的结果
• 👤 看 user 了解是否有人工干预

2.4 调度系统：超时重试、停滞升级、自动回滚

调度元数据结构

_scheduler = {
    # 配置参数
    'enabled': True,
    'stallThresholdSec': 180,         # 停滞多久后自动升级（默认180秒）
    'maxRetry': 1,                    # 自动重试次数（0=不重试，1=重试1次）
    'autoRollback': True,             # 是否自动回滚到快照

    # 运行时状态
    'retryCount': 0,                  # 当前已重试几次
    'escalationLevel': 0,             # 0=无升级 1=门下协调 2=尚书协调
    'stallSince': None,               # 何时开始停滞的时间戳
    'lastProgressAt': '2026-03-01T...',  # 最后一次获得进展的时间
    'lastEscalatedAt': '2026-03-01T...',
    'lastRetryAt': '2026-03-01T...',

    # 派发追踪
    'lastDispatchStatus': 'success',  # queued|success|failed|timeout|gateway-offline|error
    'lastDispatchAgent': 'zhongshu',
    'lastDispatchTrigger': 'state-transition',
    'lastDispatchError': '',          # 错误堆栈（如有）

    # 快照（用于自动回滚）
    'snapshot': {
        'state': 'Assigned',
        'org': '尚书省',
        'now': '等待派发...',
        'savedAt': '2026-03-01T...',
        'note': 'scheduled-check'
    }
}

调度算法

每 60 秒运行一次 handle_scheduler_scan(threshold_sec=180)：

FOR EACH 任务:
  IF state in (Done, Cancelled, Blocked): SKIP  # 终态不处理

  elapsed_since_progress = NOW - lastProgressAt
  IF elapsed_since_progress < stallThreshold: SKIP  # 最近有进展

  # ── 停滞处理逻辑 ──
  IF retryCount < maxRetry:
    ✅ 执行【重试】
    - increment retryCount；dispatch_for_state(task, new_state, trigger='taizi-scan-retry')
    - flow_log: "停滞180秒，触发自动重试第N次"

  IF escalationLevel < 2:
    ✅ 执行【升级】
    - nextLevel = escalationLevel + 1
    - target_agent = menxia (if L=1) else shangshu (if L=2)
    - wake_agent(target_agent, "💬 任务停滞，请介入协调推进")
    - flow_log: "升级至{target_agent}协调"

  IF escalationLevel >= 2 AND autoRollback:
    ✅ 执行【自动回滚】
    - restore task to snapshot.state；retryCount = 0；escalationLevel = 0
    - dispatch_for_state(task, snapshot.state, trigger='taiji-auto-rollback')
    - flow_log: "连续停滞，自动回滚到{snapshot.state}"

新后端中的停滞检测与恢复策略（orchestrator_worker.py）提供了参数化的对照实现：MAX_STALL_RETRIES = 2、MAX_ESCALATION_LEVEL = 3、STALL_RETRY_BACKOFF = [30, 60, 120]（重试退避秒数）、STALL_CHECK_INTERVAL_SEC = 60（扫描间隔）、STALL_THRESHOLD_SEC = 600（超过 10 分钟无心跳视为停滞）。_check_stalled() 每 60 秒扫描 Doing/Next 状态且 updated_at 超过阈值 10 分钟的任务，发布 task.stalled 事件；_on_task_stalled() 依次执行三层恢复策略：

1. 重试：stall_count < MAX_STALL_RETRIES 时，重新派发同一 agent（Doing/Next 按 assignee_org 查 ORG_AGENT_MAP）；
2. 升级：按 _ESCALATION_PATH 向上级流转——Doing/Next → Assigned（六部卡住退回尚书省重新派发）、Assigned → Menxia（尚书卡住退回门下复核）、Menxia → Zhongshu、Zhongshu → Taizi（orchestrator_worker.py）；
3. 标记 Blocked：重试与升级全部耗尽后，将任务置为 Blocked 并通知人工介入（orchestrator_worker.py）。

示例场景：中书省 Agent 进程崩溃，任务卡在 Zhongshu

T+0:   中书省正在规划方案；lastProgressAt = T；dispatch status = success
T+30:  Agent 进程意外崩溃（或超载无响应）；lastProgressAt 仍然 = T
T+60:  scheduler_scan：elapsed = 60 < 180，跳过
T+180: scheduler_scan：elapsed = 180 >= 180，触发处理
  ✅ 阶段1：重试 — retryCount: 0 → 1
  - dispatch_for_state('JJC-20260228-E2E', 'Zhongshu', trigger='taizi-scan-retry')
  - flow_log: "停滞180秒，自动重试第1次"
T+240: 中书省恢复（或手工重启），收到重试派发，汇报进展
  - lastProgressAt 更新为 T+240；retryCount 重置为 0；✓ 问题解决
T+360 (若仍未恢复): retryCount = 1
  ✅ 阶段2：升级 — escalationLevel: 0 → 1
  - wake_agent('menxia', "💬 任务JJC-20260228-E2E停滞，中书省无反应，请介入")
T+540 (若仍未解决):
  ✅ 阶段3：再次升级 — escalationLevel: 1 → 2
  - wake_agent('shangshu', "💬 任务长期停滞，中书省+门下省都无法推进")
T+720 (若仍未解决): escalationLevel = 2（已最大），autoRollback = true
  ✅ 阶段4：自动回滚 — snapshot.state = 'Assigned'
  - task.state: Zhongshu → Assigned；由尚书省重新派发
  - 中书省的方案保留在前一个 snapshot 版本中；用户可看到完整回滚链路

---

第三部分：核心 API 与 CLI 工具

3.1 任务操作 API 端点

对应后端实现见 edict/backend/app/api/tasks.py，新版端点采用 POST /api/tasks、POST /api/tasks/{task_id}/transition、POST /api/tasks/{task_id}/dispatch 等 RESTful 命名，并保留旧版端点（legacy.py 的 /by-legacy/{legacy_id}/...）与 GET /api/tasks/live-status、GET /api/tasks/stats 兼容接口。以下按架构文档给出的旧版端点说明语义。

任务创建：POST /api/create-task

请求：
{
  "title": "为三省六部编写完整自动化测试方案",
  "org": "中书省",           // 可选，默认太子
  "official": "中书令",      // 可选
  "priority": "normal",
  "template_id": "test_plan", // 可选
  "params": {},
  "target_dept": "兵部+刑部"  // 可选，派发建议
}

响应：
{
  "ok": true,
  "taskId": "JJC-20260228-001",
  "message": "旨意 JJC-20260228-001 已下达，正在派发给太子"
}

实现事实：create_task() 会生成 trace_id、自动计算初始 org（Task.org_for_state(initial_state, assignee_org)）、写入首条 flow_log（from: None, to: initial_state, agent: "system", reason: "任务创建"），并在同一事务中把 task.created 事件写入 outbox（task_service.py）。

任务活动流：GET /api/task-activity/{task_id}

请求：GET /api/task-activity/JJC-20260228-E2E

响应：
{
  "ok": true,
  "taskId": "JJC-20260228-E2E",
  "taskMeta": { "title": "...", "state": "Assigned", "org": "尚书省", "output": "", "block": "无", "priority": "normal" },
  "agentId": "shangshu",
  "agentLabel": "尚书省",

  "activity": [
    // flow_log (10条)
    { "at": "...", "kind": "flow", "from": "皇上", "to": "太子", "remark": "下旨：..." },
    // progress_log (11条)
    { "at": "...", "kind": "progress", "text": "已接旨。分析测试需求...", "agent": "zhongshu", "agentLabel": "中书省", "state": "Zhongshu", "org": "中书省", "tokens": 4500, "cost": 0.0045, "elapsed": 120 },
    // todos (11条)
    { "at": "...", "kind": "todos", "items": [{"id": "1", "title": "需求分析", "status": "completed"}], "agent": "zhongshu", "diff": {"changed": [{"id": "2", "from": "not-started", "to": "in-progress"}], "added": [], "removed": []} },
    // session活动 (26条总计)
    // - assistant (10条)
    { "at": "...", "kind": "assistant", "text": "基于需求，我建议采用三层测试架构...", "thinking": "💭 考虑到项目的复杂性...", "tools": [{"name": "bash", "input_preview": "find . -name '*.py' -type f | wc -l"}] },
    // - tool_result (16条)
    { "at": "...", "kind": "tool_result", "tool": "bash", "exitCode": 0, "output": "83", "durationMs": 450 }
  ],

  "activitySource": "progress+session",
  "relatedAgents": ["taizi", "zhongshu", "menxia"],
  "phaseDurations": [
    {"phase": "太子", "durationText": "30分", "ongoing": false},
    {"phase": "中书省", "durationText": "4小时32分", "ongoing": false},
    {"phase": "门下省", "durationText": "1小时15分", "ongoing": false},
    {"phase": "尚书省", "durationText": "4小时10分", "ongoing": true}
  ],
  "totalDuration": "10小时27分",
  "todosSummary": {"total": 3, "completed": 2, "inProgress": 1, "notStarted": 0, "percent": 67},
  "resourceSummary": {"totalTokens": 18500, "totalCost": 0.0187, "totalElapsedSec": 480}
}

状态推进：POST /api/advance-state/{task_id}

请求：{ "comment": "任务分明该推进了" }

响应：
{
  "ok": true,
  "message": "JJC-20260228-E2E 已推进到下一阶段 (已自动派发 Agent)",
  "oldState": "Zhongshu",
  "newState": "Menxia",
  "targetAgent": "menxia"
}

审批操作：POST /api/review-action/{task_id}

请求（准奏）：{ "action": "approve", "comment": "方案可行，已采纳改进建议" }
响应：{ "ok": true, "message": "JJC-20260228-E2E 已准奏 (已自动派发 Agent)", "state": "Assigned", "reviewRound": 1 }

请求（封驳）：{ "action": "reject", "comment": "需补充性能测试，第N轮审议" }
响应：任务返回 Zhongshu，reviewRound + 1，自动唤醒 zhongshu agent 重新规划

3.2 CLI 工具：kanban_update.py

scripts/kanban_update.py 是 Agent 与看板交互的主要入口。重要前提：脚本操作 data/tasks_source.json（JSON 看板模式）；若已部署 edict/backend（Postgres + Redis 事件总线模式），请改用后端 API 端点，或运行 edict/migration/migrate_json_to_pg.py 迁移（两种模式互相独立，数据不会自动同步）。脚本使用文件锁（file_lock.py 的 atomic_json_read/atomic_json_update）防止多 Agent 并发读写 JSON 文件。

除架构文档列出的 7 个命令外，实际脚本还包含 block、confirm、todo、memory、task-memo、shared-memo、delegate、delegate-result 等扩展命令（命令最小参数表见 kanban_update.py）。以下为核心命令：

命令 1：创建任务（太子或中书手工）

python3 scripts/kanban_update.py create \
  JJC-20260228-E2E \
  "为三省六部编写完整自动化测试方案" \
  Zhongshu \
  中书省 \
  中书令

# 说明：通常不需要手工运行（看板API自动触发），除非 debug

创建时会做旨意标题校验（_is_valid_task_title）：标题少于 6 字、命中垃圾标题集合（好的/收到/知道了/测试 等）、纯标点、形似文件路径都会被拒绝（kanban_update.py）。同时 _sanitize_text 会剥离 Conversation 元数据、代码块、文件路径、URL、传旨/下旨前缀等噪声（kanban_update.py）。已完结（Done/Cancelled）任务不可被覆盖。

命令 2：更新状态

python3 scripts/kanban_update.py state \
  JJC-20260228-E2E \
  Menxia \
  "方案提交门下省审议"

# - 第一个参数：task_id；第二个参数：新状态（Pending/Taizi/Zhongshu/...）
# - 第三个参数：可选，描述信息（记录到 now 字段）
# 效果：task.state = Menxia；task.org 自动推断为 "门下省"；触发派发 menxia agent；flow_log 记录转移

state 命令内置流转合法性校验（对照 _VALID_TRANSITIONS），非法转换直接拒绝并写审计日志；对于高风险转换（Review→Done 需门下确认、Doing→Cancelled 需尚书确认、Menxia→Cancelled 需中书确认），会先进入 PendingConfirm 中间态，由 confirm 命令（权限方 CONFIRM_AUTHORITY 分别为 menxia/shangshu/zhongshu）批准或驳回（kanban_update.py）。

命令 3：添加流转记录

python3 scripts/kanban_update.py flow \
  JJC-20260228-E2E \
  "中书省" \
  "门下省" \
  "📋 方案提交审核，请审议"

# 参数1：task_id；参数2：from_dept（谁在上报）；参数3：to_dept（流转到谁）；参数4：remark
# 注意：只记录 flow_log，不改变 task.state（多用于部门间协调的细节流转）

命令 4：实时进展汇报（重点！）

python3 scripts/kanban_update.py progress \
  JJC-20260228-E2E \
  "已完成需求分析和方案初稿，现正征询工部意见" \
  "1.需求分析✅|2.方案设计✅|3.工部咨询🔄|4.待门下审议"

# 参数1：task_id；参数2：进展文本说明；参数3：todos 快照（| 分隔，emoji 标记状态）
# todos 解析规则：以 ✅ 结尾 → completed；以 🔄 结尾 → in-progress；其他 → not-started

效果：progress_log 添加新条目（at/agent/agentLabel/text/state/org/todos），可附加 --tokens、--cost、--elapsed 上报资源消耗；看板即时渲染活动条目、更新 todos 进度条、累加资源消耗。progress_log 上限 100 条（MAX_PROGRESS_LOG），超出自动裁剪。

命令 5：任务完成

python3 scripts/kanban_update.py done \
  JJC-20260228-E2E \
  "https://github.com/org/repo/tree/feature/auto-test" \
  "自动化测试方案已完成，涵盖单元/集成/E2E三层，通过率98.5%"

# 参数1：task_id；参数2：output URL；参数3：最终总结
# 效果：task.state = Done（从 Review 推进）；task.output 写入；自动发送飞书消息给皇上（太子转报）

实现细节：done 命令只允许在 Doing/Next 状态执行；若 todos 未全部 completed 则拒绝收口（"todos 未完成（x/y），禁止直接收口"）；完成后状态进入 Review（尚书省汇总审查）而非直接 Done（kanban_update.py）。若 output 是本地文件路径，会自动生成 outputMeta（存在性 + 最后修改时间）。

命令 6-8：停止 / 恢复 / 取消

# 叫停（随时可恢复）
python3 scripts/kanban_update.py stop JJC-20260228-E2E "等待工部反馈继续"
# - task.state 暂存（_prev_state）；task.block 写入；看板显示 "⏸️ 已叫停"

# 恢复
python3 scripts/kanban_update.py resume JJC-20260228-E2E "工部已反馈，继续执行"
# - task.state 恢复到 _prev_state；重新派发 agent

# 取消（不可恢复）
python3 scripts/kanban_update.py cancel JJC-20260228-E2E "业务需求变更，任务作废"
# - task.state = Cancelled；flow_log 记录取消原因

扩展命令：todo / block / confirm

# 添加或更新子任务（约束：同一时刻最多只有 1 个 in-progress 状态的 todo）
python3 scripts/kanban_update.py todo JJC-20260228-E2E 1 "实现API接口" in-progress
python3 scripts/kanban_update.py todo JJC-20260228-E2E 1 "" completed --detail "POST /api/tasks 已实现"

# 标记阻塞（可恢复）
python3 scripts/kanban_update.py block JJC-20260228-E2E "等待工部反馈"

# 确认/驳回 PendingConfirm 状态的高风险操作
python3 scripts/kanban_update.py confirm JJC-20260228-E2E approve "确认完结"
python3 scripts/kanban_update.py confirm JJC-20260228-E2E reject "还需补充"

扩展能力：三级记忆与委派

# Agent 永久记忆（feedback/experience/preference + 相关性标签）
python3 scripts/kanban_update.py memory zhongshu experience "门下省偏好结构化方案" --source-task JJC-20260228-E2E --tags "审议,方案"

# 任务上下文记忆（跨 Agent 传递决策链）
python3 scripts/kanban_update.py task-memo JJC-20260228-E2E zhongshu "三层测试架构,优先pytest" --warnings "注意E2E环境依赖"

# 全局共享记忆（所有 Agent 可读规则）
python3 scripts/kanban_update.py shared-memo "所有产出必须附带测试" shangshu

# 子 Agent 无状态委派（防死锁：深度≤3 层 + 循环委派拒绝）
python3 scripts/kanban_update.py delegate JJC-20260228-E2E shangshu libu "整理文档排版" "返回文档路径"
python3 scripts/kanban_update.py delegate-result JJC-20260228-E2E-sub-xxxx "文档已完成：/docs/xxx.md"

---

第四部分：对标与对比 — 制度化协作 vs 自由讨论

CrewAI / AutoGen 的传统方式 vs 三省六部的制度化方式

维度	CrewAI	AutoGen	三省六部
协作模式	自由讨论（Agent 自主选择协作对象）	面板+回调（Human-in-the-loop）	制度化协作（权限矩阵+状态机）
质量保障	依赖 Agent 智能（无审核）	Human 审核（频繁中断）	自动审核（门下省必审）+可干预
权限控制	❌ 无	⚠️ Hard-coded	✅ 配置化权限矩阵
可观测性	低（Agent 消息黑盒）	中（Human 看到对话）	极高（59条活动/任务）
可干预性	❌ 无（跑起来后很难叫停）	✅ 有（需要人工批准）	✅ 有（一键stop/cancel/advance）
任务分发	不确定（Agent 自主选）	确定（Human 手工分）	自动确定（权限矩阵+状态机）
吞吐量	1 任务 1 Agent（串行讨论）	1 任务 1 Team（需人工管理）	多任务并行（六部同时执行）
失败恢复	❌（重新开始）	⚠️（需人工调试）	✅（自动重试3阶段）
成本控制	不透明（没有成本上限）	中等（Human 可叫停）	透明（每条 progress 上报成本）

业务契约的严格性

CrewAI 的"温和"方式：Agent 可以自由选择下一步工作，可能发错人、可能重复。

if task_seems_done:
    send_message_to_someone()  # 可能发错人，可能重复

三省六部的"严格"方式：任务状态严格受限，下一步由系统决定；状态转移只能通过 API，不能绕过；越权调用会被权限检查拦截。

if task.state == 'Zhongshu' and agent_id == 'zhongshu':
    deliver_plan_to_menxia()   # 中书不能直接转尚书，必须经过门下审议
    try:
        dispatch_to(shangshu)  # ❌ 权限检查拦截
    except PermissionError:
        log.error(f'zhongshu 无权越权调用 shangshu')

---

第五部分：故障场景与恢复机制

场景 1：Agent 进程崩溃

症状：任务卡在某个状态，180秒无新进展；报警：太子调度系统检测到停滞
自动处理流程：
  T+0:   崩溃
  T+180: scan 检测到停滞
    ✅ 第1阶段：自动重试 — 派发消息到 agent（唤醒或重启），如果恢复则继续
  T+360: 若仍未恢复
    ✅ 第2阶段：升级协调 — 唤醒门下省 agent，汇报 "中书省无响应，请介入"，门下可能接管或代理工作
  T+540: 若仍未恢复
    ✅ 第3阶段：再次升级 — 唤醒尚书省 agent，企业级协调
  T+720: 若仍未恢复
    ✅ 第4阶段：自动回滚 — 恢复到前一个稳定状态，派发给尚书省重新处理

场景 2：Agent 作恶（伪造数据）

假设 zhongshu agent 想骗过系统（直接在 JSON 里伪造门下省的准奏记录），系统防御如下：

# 1. 权限验证：API 层检查 HTTP 请求者身份
#    ├─ 来自 zhongshu agent 的请求不能直接 flow
#    ├─ 必须通过 flow_log 记录，且签名验证；签名不匹配则拒绝
# 2. 状态机验证：状态转移受控
#    ├─ 即使 flow_log 被篡改，state 仍然是 Zhongshu
#    ├─ 下一步只能由 gate-keeper 系统转移；zhongshu 无权自己改 state
# 结果：❌ Agent 的伪造被系统拦截

此外，新后端的 DispatchWorker 还内置了 Prompt 注入检测：_sanitize_agent_output 用 _INJECTION_PATTERNS（如"忽略指令/规则"、"ignore instructions"、system:、伪装管理员、"override/bypass/skip review"等正则）扫描 Agent 输出，命中即发布 agent.injection.detected 告警事件（dispatch_worker.py）。

场景 3：业务流程违反（如中书越权调尚书改方案）

# 中书省想绕过门下审议，直接咨询尚书省
try:
    result = dispatch_to_agent('shangshu', '请帮我审查一下这个方案')
except PermissionError:
    log.error('zhongshu 无权调用 shangshu (仅限: menxia, shangshu)')

# 门下省想升级到皇上
try:
    result = dispatch_to_agent('taizi', '我需要皇上的指示')
except PermissionError:
    log.error('menxia 无权调用 taizi')

---

第六部分：监控与可观测性

看板的 10 个视图面板

1. 全任务列表 — 按创建时间倒序，快速过滤：活跃/完成/已封驳
2. 按状态分类 — Pending/Taizi/Zhongshu/Menxia/Assigned/Doing/Review/Done/Cancelled
3. 按部门分类 — 太子/中书省/门下省/尚书省/六部（并行视图）/已派发
4. 按优先级分类 — 🔴 Critical / 🟠 High / 🟡 Normal / 🔵 Low
5. Agent 在线状态
   ├─ 🟢 运行中（正在处理任务） ├─ 🟡 待命（最近有活动，闲置）
   ├─ ⚪ 空闲（超过10分钟无活动）├─ 🔴 离线（Gateway 未启动）
   └─ ❌ 未配置（工作空间不存在）
6. 任务详情面板 — 基本信息、完整活动流、阶段耗时统计、Todos 进度条、资源消耗
7. 停滞任务监控 — 列出超阈值任务，显示停滞时长，快速操作：重试/升级/回滚
8. 审批工单池 — 所有 Menxia 等待审批任务，按停留时长排序，一键准奏/封驳
9. 今日概览 — 新建数/完成数/平均流转时长/各 Agent 活动频率
10. 历史报表 — 周报（人均产出、平均周期）/月报（部门协作效率）/成本分析

实时 API：Agent 在线检测

GET /api/agents-status

响应：
{
  "ok": true,
  "gateway": { "alive": true, "probe": true, "status": "🟢 运行中" },
  "agents": [
    {
      "id": "taizi", "label": "太子",
      "status": "running",        // running|idle|offline|unconfigured
      "statusLabel": "🟢 运行中",
      "lastActive": "03-02 14:30", "lastActiveTs": 1708943400000,
      "sessions": 42, "hasWorkspace": true, "processAlive": true
    }
    // ... 其他 agent ...
  ]
}

---

第七部分：使用示例与最佳实践

完整案例：创建 → 分发 → 执行 → 完成

# ═══════════════════════════════════════════════════════════
# 第1步：皇上下旨（飞书消息或看板API）
# ═══════════════════════════════════════════════════════════
curl -X POST http://127.0.0.1:7891/api/create-task \
  -H "Content-Type: application/json" \
  -d '{ "title": "编写三省六部协议文档", "priority": "high" }'
# 响应：JJC-20260302-001 已创建；太子Agent 收到通知 "📜 皇上旨意..."

# ═══════════════════════════════════════════════════════════
# 第2步：太子接旨分拣（Agent自动）
# ═══════════════════════════════════════════════════════════
python3 scripts/kanban_update.py state JJC-20260302-001 Zhongshu "分拣完毕，转中书省起草"
# 中书省Agent 收到派发通知

# ═══════════════════════════════════════════════════════════
# 第3步：中书起草（Agent工作）
# ═══════════════════════════════════════════════════════════
# 第一次汇报（30分钟后）：
python3 scripts/kanban_update.py progress JJC-20260302-001 \
  "已完成需求分析，拟定三部分文档：概述|技术栈|使用指南" \
  "1.需求分析✅|2.文档规划✅|3.内容编写🔄|4.审查待完成"
# 看板显示：进度条 50%；消耗 1200 tokens, $0.0012, 18分钟

# 第二次汇报（再过90分钟）：
python3 scripts/kanban_update.py progress JJC-20260302-001 \
  "文档初稿已完成，现提交门下省审议" "1.需求分析✅|2.文档规划✅|3.内容编写✅|4.待审查"
python3 scripts/kanban_update.py flow JJC-20260302-001 "中书省" "门下省" "提交审议"
python3 scripts/kanban_update.py state JJC-20260302-001 Menxia "方案提交门下省审议"

# ═══════════════════════════════════════════════════════════
# 第4步：门下审议（Agent工作）
# ═══════════════════════════════════════════════════════════
# 情景A：准奏
python3 scripts/kanban_update.py state JJC-20260302-001 Assigned "✅ 准奏，已采纳改进建议"
python3 scripts/kanban_update.py flow JJC-20260302-001 "门下省" "尚书省" "✅ 准奏：文档质量良好，建议补充代码示例"
# 情景B：封驳
python3 scripts/kanban_update.py state JJC-20260302-001 Zhongshu "🚫 封驳：需补充协议规范部分"
python3 scripts/kanban_update.py flow JJC-20260302-001 "门下省" "中书省" "🚫 封驳：协议部分过于简略，需补充权限矩阵示例"

# ═══════════════════════════════════════════════════════════
# 第5步：尚书派发（Agent工作）
# ═══════════════════════════════════════════════════════════
# 尚书分析派发对象：礼部（排版）、兵部（代码示例）、工部（部署文档）
python3 scripts/kanban_update.py state JJC-20260302-001 Doing "派发给礼部+兵部+工部三部并行执行"
python3 scripts/kanban_update.py flow JJC-20260302-001 "尚书省" "六部" "派发执行：礼部排版|兵部代码示例|工部基础设施部分"

# ═══════════════════════════════════════════════════════════
# 第6步：六部执行（并行，各自 progress 汇报）
# ═══════════════════════════════════════════════════════════
python3 scripts/kanban_update.py progress JJC-20260302-001 "已完成文档排版和目录调整" "1.排版✅|2.目录调整✅|3.等待代码示例|4.等待基础设施部分"
python3 scripts/kanban_update.py progress JJC-20260302-001 "已编写5个代码示例（权限检查、派发流程、session融合等）" "1.分析需求✅|2.编码示例✅|3.集成文档🔄|4.测试验证"
python3 scripts/kanban_update.py progress JJC-20260302-001 "已编写Docker+K8s部署部分" "1.Docker编写✅|2.K8s配置✅|3.一键部署脚本🔄|4.部署文档待完成"

# ═══════════════════════════════════════════════════════════
# 第7步：尚书汇总（Agent工作）
# ═══════════════════════════════════════════════════════════
python3 scripts/kanban_update.py progress JJC-20260302-001 \
  "全部部门已完成。汇总成果：文档已排版9章、15个代码示例、完整部署指南" "1.排版✅|2.代码示例✅|3.基础设施✅|4.汇总✅"
python3 scripts/kanban_update.py state JJC-20260302-001 Review "所有部门完成，进入审查阶段"

# ═══════════════════════════════════════════════════════════
# 第8步：完成（终态）
# ═══════════════════════════════════════════════════════════
python3 scripts/kanban_update.py done JJC-20260302-001 \
  "https://github.com/org/repo/docs/architecture.md" \
  "三省六部协议文档已完成，包含89页，5个阶段历时3天，总消耗成本$2.34"
# 看板显示：Done ✅；总耗时 3天2小时45分；完整活动流 79条；87500 tokens, $2.34

# 查询最终成果
curl http://127.0.0.1:7891/api/task-activity/JJC-20260302-001
# 响应：taskMeta.state=Done；activity=79条完整流转链；resourceSummary 汇总

最佳实践建议

1. 进展汇报要频繁：每 30 分钟或关键节点用 progress 上报一次，附带 todos 快照与 --tokens/--cost/--elapsed，既是可观测性来源，也是调度系统判断"是否停滞"的依据（lastProgressAt 更新）；
2. 六部只用 progress/todo/done/block：执行类 Agent 不要直接改 state，完成回报交给 done（自动进入 Review）或委派结果 delegate-result；
3. todos 保持单一 in-progress：同一时刻只有一个进行中的子任务，避免并行子任务状态混乱，也保证 ready_to_close 信号准确；
4. 高风险操作走 confirm 流程：完结、取消等不可逆操作会进入 PendingConfirm，由权限方确认后再执行；
5. 用三级记忆沉淀经验：memory/task-memo/shared-memo 让跨 Agent 决策链可追溯，并在下次派发时自动注入相关记忆。

---

补充概念：Workflow state vs execution ownership

在 Edict 工作流中，**workflow state（工作流状态）**与 **execution ownership（执行归属）**相关但并不总是相同。

• Workflow state 描述任务当前所处的制度阶段（Taizi / Zhongshu / Menxia / Assigned / Doing / Review / Done 等），这是看板/UI 上展示的阶段，反映任务在既定流程中的位置；
• Execution ownership 指当前实际持有下一步动作、正在写下一条 progress/日志更新的运行时 agent/session。

在简单流程中两者看起来一致；在复杂多阶段流程中它们可能暂时分离——例如 workflow state 已经显示 Assigned，而活跃的运行时执行上下文仍挂在早期阶段的派发链上。

这个区分对以下场景尤其有用：

• 调试交接（handoff）行为；
• 理解为什么在完全隔离的执行交接发生之前就出现了后期阶段的标签；
• 思考未来的改进方向（如 stage-isolated orchestration，阶段隔离编排）。

简言之：workflow state = 任务在制度流程中的位置；execution ownership = 哪个运行时路径当前持有下一步动作。此概念澄清不改变运行时语义，仅用于指导高级真实场景的使用。

---

总结

三省六部是一个制度化的 AI 多 Agent 系统，而非传统的"自由讨论"框架。它通过：

1. 业务层：模仿古代帝国官僚体系，建立分权制衡的组织结构（皇上 → 太子 → 中书 → 门下 → 尚书 → 六部）；
2. 技术层：状态机（STATE_TRANSITIONS/STATE_AGENT_MAP，见 edict/backend/app/models/task.py）+ 配置化权限矩阵（agents.json + AGENT_POLICY）+ 事件驱动自动派发（orchestrator_worker.py、dispatch_worker.py）+ 停滞重试/升级/回滚调度，确保流程可控；
3. 观测层：React 看板 10 个面板 + 完整活动流（flow_log + progress_log + session JSONL 融合，数十条/任务），实时掌握全局；
4. 介入层：一键 stop/cancel/advance/resume + PendingConfirm 确认机制，遇到异常能立即纠正。

核心价值：用制度确保质量，用透明确保信心，用自动化确保效率。相比 CrewAI/AutoGen 的"自由+人工管理"，三省六部提供了一套企业级、可治理、可审计的多 Agent 协作框架，其业务契约、权限矩阵与调度恢复机制均可直接对照本仓库源码继续深入研读与二次开发。