Coordinator 通用模式

本 skill 定义了所有 coordinator agents 共享的通用模式，包括：

Phase 验证逻辑
错误处理模式
TodoWrite 管理
状态说明

Phase 验证逻辑

所有 master-coordinator 必须在初始化时验证 phase 参数：

def validate_phase(phase_arg, valid_phases): """ 验证 phase 参数

Args:
    phase_arg: 用户传入的 phase 参数（如 "0,1,2" 或 "all"）
    valid_phases: 有效 phase 列表（如 ["0", "1", "2", "3", "4", "5", "all"]）

Returns:
    (is_valid, result): 如果有效，result 是 phase 列表；如果无效，result 是错误响应
"""
if phase_arg == "all":
    # 返回所有数字 phases（排除 "all"）
    return True, sorted([p for p in valid_phases if p != "all"], key=int)

phases = phase_arg.split(",")
invalid_phases = [p for p in phases if p not in valid_phases]

if invalid_phases:
    return False, {
        "status": "failed",
        "error": {
            "code": "INVALID_PHASE",
            "message": f"无效的 phase 参数: {invalid_phases}",
            "valid_values": valid_phases,
            "received": phase_arg,
            "suggestion": f"有效值: 0-{len(valid_phases)-2} 的数字或 'all'，多个用逗号分隔"
        }
    }

return True, sorted(set(phases), key=int)

各工作流的有效 Phases

工作流有效 Phases 说明

Bugfix 0-5 6 阶段

PR Review 0-7 8 阶段

CI Job 0-6 7 阶段

Execute Plan 0-5 6 阶段

错误处理模式

所有 coordinator 必须处理以下错误类型：

JSON 解析错误

当 agent 返回的内容无法解析为有效 JSON 时：

try: result = json.loads(agent_output) except json.JSONDecodeError as e: return { "status": "failed", "error": { "code": "JSON_PARSE_ERROR", "message": "Agent 输出无法解析为 JSON", "phase": current_phase, "agent": agent_name, "parse_error": str(e), "raw_output_preview": agent_output[:500], # 前 500 字符供调试 "suggestion": "检查 agent 是否正确返回 JSON 格式，或重试命令" } }

Agent 执行超时

if agent_result.error.code == "TIMEOUT": return { "status": "failed", "error": { "code": "AGENT_TIMEOUT", "message": f"Agent {agent_name} 执行超时", "phase": current_phase, "timeout_ms": agent_result.error.timeout_ms, "suggestion": "任务可能过于复杂，建议拆分或简化输入" } }

响应截断

当 agent 输出超过长度限制被截断时：

if agent_result.truncated: # 记录警告但尝试继续 warnings.append({ "code": "OUTPUT_TRUNCATED", "message": f"Agent {agent_name} 输出被截断", "original_length": agent_result.original_length, "truncated_length": agent_result.truncated_length, "impact": "可能丢失部分诊断信息" }) # 如果关键字段缺失，则停止 if not validate_required_fields(agent_result): return { "status": "failed", "error": { "code": "TRUNCATION_DATA_LOSS", "message": "输出截断导致关键数据丢失", "missing_fields": get_missing_fields(agent_result), "suggestion": "请简化输入或分批处理" } }

用户取消

if user_choice in ["取消", "停止"]: return { "status": "user_cancelled", "phase": current_phase, "reason": "用户选择停止执行", "completed_work": {...} # 已完成的工作 }

Agent 调用失败（通用）

if agent_result.status == "failed": return { "status": "failed", "error": { "phase": current_phase, "agent": agent_name, "code": agent_result.error.code, "message": agent_result.error.message } }

错误恢复机制（可选）

对于支持错误恢复的 coordinator：

可恢复错误类型

RECOVERABLE_ERRORS = { "TIMEOUT": True, # 超时可重试 "RATE_LIMIT": True, # 限流可重试 "OUTPUT_TRUNCATED": True, # 截断可简化输入重试 }

MAX_RETRIES = 2 # 最多重试 2 次

def is_recoverable(error): """判断错误是否可恢复""" return RECOVERABLE_ERRORS.get(error.code, False)

TodoWrite 管理

所有 coordinator 必须使用 TodoWrite 跟踪执行进度：

初始化 Todo 列表

def create_phase_todos(phases, phase_descriptions): """ 创建 Phase 任务列表

Args:
    phases: Phase 列表（如 ["0", "1", "2"]）
    phase_descriptions: Phase 描述映射（如 {"0": ("问题收集", "收集中"), ...}）

Returns:
    todos 列表
"""
todos = []
for i, phase in enumerate(phases):
    desc, active_form = phase_descriptions.get(phase, (f"Phase {phase}", f"执行 Phase {phase}"))
    todos.append({
        "content": f"Phase {phase}: {desc}",
        "status": "in_progress" if i == 0 else "pending",
        "activeForm": active_form
    })
return todos

更新 Todo 状态

def on_phase_complete(todos, phase_index): """完成 Phase 后更新状态""" todos[phase_index]["status"] = "completed" if phase_index + 1 < len(todos): todos[phase_index + 1]["status"] = "in_progress" return todos

各工作流的 Phase 描述

Bugfix 工作流：

BUGFIX_PHASES = { "0": ("问题收集与分类", "收集中"), "1": ("诊断分析", "分析中"), "2": ("方案设计", "设计中"), "3": ("方案文档化", "文档化中"), "4": ("实施执行", "执行中"), "5": ("验证与审查", "审查中") }

PR Review 工作流：

PR_REVIEW_PHASES = { "0": ("初始化", "初始化中"), "1": ("评论获取", "获取评论中"), "2": ("评论过滤", "过滤评论中"), "3": ("评论分类", "分类评论中"), "4": ("修复协调", "协调修复中"), "5": ("回复生成", "生成回复中"), "6": ("回复提交", "提交回复中"), "7": ("审查与汇总", "审查汇总中") }

CI Job 工作流：

CI_JOB_PHASES = { "0": ("初始化", "初始化中"), "1": ("日志获取", "获取日志中"), "2": ("失败分类", "分类失败中"), "3": ("根因分析", "分析根因中"), "4": ("修复执行", "执行修复中"), "5": ("验证与审查", "验证审查中"), "6": ("汇总报告", "生成报告中") }

Execute Plan 工作流：

EXECUTE_PLAN_PHASES = { "0": ("初始化与计划解析", "初始化中"), "1": ("计划验证", "验证中"), "2": ("方案细化", "细化方案中"), "3": ("批次执行", "执行中"), "4": ("Review 审查", "审查中"), "5": ("汇总报告", "生成报告中") }

状态说明

所有 coordinator 输出的 status 字段使用统一的语义：

status 含义适用场景

success

所有 Phase 成功完成正常完成

failed

某个 Phase 失败且无法继续不可恢复错误

partial

部分任务失败，但流程完成有剩余问题

user_cancelled

用户选择停止用户主动取消

dry_run_complete

Dry run 模式完成分析 --dry-run 模式

必填输出字段

每个 coordinator 的 JSON 输出必须包含：

{ "status": "success|failed|partial|user_cancelled|dry_run_complete", "agent": "xxx-master-coordinator", "phases_completed": ["phase_0", "phase_1", ...], "errors": [], "warnings": [] }

关键原则

闭环执行：所有逻辑在 agent 内部完成，不依赖命令层
状态透明：每个 Phase 的输出都保存并传递到下一 Phase
用户控制：关键决策点使用 AskUserQuestion 询问用户
进度可见：使用 TodoWrite 让用户了解执行进度
错误隔离：单个任务失败不应影响其他独立任务

coordinator-patterns

Safety Notice

Copy this and send it to your AI assistant to learn

可恢复错误类型

Source Transparency

Related Skills

bugfix-workflow

elements-of-style

frontend-bugfix

pr-review-analysis