Supervisor Pattern 深度指南：任务分解质量与工作流级错误恢复

def check_decomposition_quality(tasks):
    """
    检查分解质量

    每个任务应该能回答：
    1. 这个任务的输入是什么？（来自 Supervisor 还是其他 Worker）
    2. 这个任务的输出是什么？
    3. 其他任务是否依赖这个任务的输出？
    """
    issues = []

    for task in tasks:
        # 检查1：输入是否已就绪
        for dep in task.dependencies:
            if not dep.is_produced_before(task):
                issues.append(f"[{task.id}] 依赖 {dep.id}，但该任务的执行时机无法保证在 {task.id} 之前")

        # 检查2：任务是否太简单（过度分解）
        if task.complexity < COMPLEXITY_THRESHOLD:
            # 再检查是否值得单独成为一个任务
            if not task.has_side_effects:
                issues.append(f"[{task.id}] 过于简单，考虑合并到其他任务")

        # 检查3：任务是否太复杂（欠分解）
        if task.complexity > COMPLEXITY_THRESHOLD * 3:
            issues.append(f"[{task.id}] 过于复杂，考虑拆分为多个子任务")

    return issues

def check_dependency_graph(tasks):
    """
    依赖关系必须是 DAG（有向无环图）
    - 每条边从依赖方指向被依赖方
    - 不能有环
    """
    # 构建邻接表
    graph = {task.id: [] for task in tasks}
    in_degree = {task.id: 0 for task in tasks}

    for task in tasks:
        for dep in task.dependencies:
            graph[dep.id].append(task.id)
            in_degree[task.id] += 1

    # 检测环（Kahn 算法）
    queue = [t for t in tasks if in_degree[t.id] == 0]
    processed = 0

    while queue:
        node = queue.pop(0)
        processed += 1
        for neighbor in graph[node.id]:
            in_degree[neighbor] -= 1
            if in_degree[neighbor] == 0:
                queue.append(neighbor)

    if processed != len(tasks):
        # 有环
        cycle_nodes = [t for t in tasks if in_degree[t.id] > 0]
        raise CycleDetectedError(f"检测到循环依赖: {cycle_nodes}")

    return True

from enum import Enum
from typing import Callable
import time

class ErrorType(Enum):
    TRANSIENT = "transient"       # 瞬时错误
    CAPABILITY = "capability"    # 能力不足
    DEPENDENCY = "dependency"    # 依赖失败
    RESOURCE = "resource"        # 资源耗尽
    UNKNOWN = "unknown"           # 未知错误

class RecoveryStrategy:
    def __init__(self):
        self.strategies: dict[ErrorType, Callable] = {
            ErrorType.TRANSIENT: self._handle_transient,
            ErrorType.CAPABILITY: self._handle_capability,
            ErrorType.DEPENDENCY: self._handle_dependency,
            ErrorType.RESOURCE: self._handle_resource,
            ErrorType.UNKNOWN: self._handle_unknown,
        }

    def classify(self, error_msg: str, context: dict) -> ErrorType:
        """根据错误信息和上下文分类错误类型"""
        error_lower = error_msg.lower()

        # 瞬时错误信号
        transient_signals = ["timeout", "connection", "permission denied", "econnreset", "etimedout"]
        if any(signal in error_lower for signal in transient_signals):
            return ErrorType.TRANSIENT

        # 能力不足信号
        capability_signals = ["context.*overflow", "unknown.*tool", "malformed.*output", "does not understand"]
        if any(signal in error_lower for signal in capability_signals):
            return ErrorType.CAPABILITY

        # 依赖失败信号
        dependency_signals = ["not found", "undefined", "not.*available", "upstream"]
        if any(signal in error_lower for signal in dependency_signals):
            return ErrorType.DEPENDENCY

        # 资源耗尽信号
        resource_signals = ["max_tokens", "out of memory", "budget"]
        if any(signal in error_lower for signal in resource_signals):
            return ErrorType.RESOURCE

        return ErrorType.UNKNOWN

    def recover(self, error: Exception, task: dict, context: dict) -> dict:
        """根据错误类型执行对应的恢复策略"""
        error_type = self.classify(str(error), context)
        strategy = self.strategies[error_type]

        # 记录恢复尝试
        recovery_log = {
            "error": str(error),
            "error_type": error_type.value,
            "task_id": task.get("id"),
            "attempt": context.get("retry_count", 0),
            "timestamp": now_iso(),
        }

        result = strategy(task, context, error)
        recovery_log["result"] = result

        return recovery_log

    def _handle_transient(self, task: dict, context: dict, error: Exception) -> dict:
        """瞬时错误：指数退避重试"""
        retry_count = context.get("retry_count", 0)
        max_retries = 3

        if retry_count >= max_retries:
            return {"action": "escalate", "reason": "max_retries_exceeded"}

        # 指数退避：1s, 2s, 4s
        wait_seconds = 2 ** retry_count
        time.sleep(wait_seconds)

        return {"action": "retry", "wait_seconds": wait_seconds, "task": task}

    def _handle_capability(self, task: dict, context: dict, error: Exception) -> dict:
        """能力不足：检查是否欠分解，或升级 Worker"""
        task_complexity = context.get("task_complexity", "unknown")

        if task_complexity == "over_decomposed":
            # 过度分解 → 合并相关任务
            return {"action": "merge_tasks", "tasks_to_merge": context.get("related_tasks", [])}

        elif task_complexity == "under_decomposed":
            # 欠分解 → 拆分为更小的子任务
            return {"action": "refine_task", "original_task": task}

        else:
            # Worker 能力不足 → 升级
            return {"action": "upgrade_worker", "task": task}

    def _handle_dependency(self, task: dict, context: dict, error: Exception) -> dict:
        """依赖失败：重新拓扑排序"""
        dag = context.get("dependency_graph")

        if dag.has_cycle():
            return {"action": "escalate", "reason": "circular_dependency_detected"}

        # 重新排序
        new_order = dag.topological_sort()
        return {"action": "reschedule", "new_order": new_order}

    def _handle_resource(self, task: dict, context: dict, error: Exception) -> dict:
        """资源耗尽：压缩 context 或拆分任务"""
        resource_type = context.get("resource_type", "unknown")

        if resource_type == "context":
            # Context 溢出 → 摘要清理后重试
            return {"action": "compress_context", "task": task}
        else:
            # 其他资源 → 拆分任务
            return {"action": "split_task", "task": task}

    def _handle_unknown(self, task: dict, context: dict, error: Exception) -> dict:
        """未知错误：上报人类"""
        return {
            "action": "escalate",
            "reason": "unknown_error",
            "error": str(error),
            "task": task,
            "context_summary": summarize_context(context),
        }

# Checkpoint 策略
CHECKPOINT_AFTER = [
    "task_type: architecture_decision",  # 架构决策影响后续所有实现
    "task_type: refactoring",            # 重构改变多个文件，容易引发连锁反应
    "task_type: schema_change",          # Schema 变更影响上下游
    "status: needs_human_review",        # 需要人工审核的节点
]

def should_checkpoint(task: dict) -> bool:
    """判断任务完成后是否需要 checkpoint"""
    return (
        task.get("type") in CHECKPOINT_AFTER
        or task.get("status") == "needs_human_review"
        or task.get("risk_level") == "high"
    )

def recover_workflow(failed_task_id: str, checkpoint_manager: CheckpointManager):
    """
    增量恢复：只重做失败任务及其下游
    1. 找到失败任务的上一个有效 Checkpoint
    2. 恢复状态到 Checkpoint
    3. 只重新执行 Checkpoint 之后的任务（而不是整个工作流）
    """
    # Step 1: 找到最近的有效 Checkpoint
    last_valid_checkpoint = checkpoint_manager.get_last_valid(failed_task_id)

    # Step 2: 恢复状态
    restored_state = checkpoint_manager.restore(last_valid_checkpoint)

    # Step 3: 确定需要重做的任务范围
    # 只重做：checkpoint 之后的任务 + 失败任务的下游
    tasks_to_rerun = get_downstream_tasks(last_valid_checkpoint.task_id)
    tasks_to_rerun.add(failed_task_id)

    # Step 4: 按依赖顺序重做
    for task in topological_sort(tasks_to_rerun):
        result = execute_task(task, restored_state)
        if result.is_failed():
            # 如果重做仍然失败，说明问题不在下游，可能是 Checkpoint 之前的问题
            # 递归回退到更早的 Checkpoint
            return recover_workflow(task.id, checkpoint_manager)
        restored_state.update(task.id, result)

    return restored_state

class SupervisorHealthMonitor:
    def __init__(self):
        self.metrics = {
            "task_completion_rate": [],   # 任务完成率
            "retry_rate": [],              # 重试率（过高说明分解有问题）
            "escalation_rate": [],         # 上报率（过高说明 Worker 不匹配）
            "avg_task_duration": [],       # 平均任务时长
            "context_contamination": 0,    # 上下文污染事件数
        }

    def record_task_outcome(self, task_id: str, outcome: dict):
        """记录任务结果，用于计算健康度"""
        self.metrics["task_completion_rate"].append(
            1.0 if outcome["status"] == "success" else 0.0
        )
        self.metrics["retry_rate"].append(outcome.get("retry_count", 0))
        self.metrics["escalation_rate"].append(
            1.0 if outcome["status"] == "escalated" else 0.0
        )
        self.metrics["avg_task_duration"].append(outcome.get("duration_seconds", 0))

    def compute_health_score(self) -> float:
        """
        计算工作流健康度得分（0-100）

        权重：
          完成率（30%）：任务是否顺利完成
          重试率（25%）：重试次数是否异常高
          上报率（20%）：需要人工介入的比例
          时效性（15%）：任务执行时间是否正常
          稳定性（10%）：上下文污染事件
        """
        completion_rate = mean(self.metrics["task_completion_rate"])
        retry_rate = normalize_retry_rate(mean(self.metrics["retry_rate"]))
        escalation_rate = mean(self.metrics["escalation_rate"])
        duration_anomaly = self._detect_duration_anomaly()

        score = (
            completion_rate * 0.30
            + retry_rate * 0.25
            + (1 - escalation_rate) * 0.20
            + duration_anomaly * 0.15
            + self._compute_stability() * 0.10
        )

        return round(score * 100, 1)

    def should_self_adjust(self) -> bool:
        """判断是否需要自我调整"""
        score = self.compute_health_score()
        return score < 70  # 健康度低于 70 时触发调整

def self_adjust(supervisor: Supervisor, health_report: dict):
    """根据健康度报告自动调整"""
    actions_taken = []

    if health_report["retry_rate"] > 0.5:
        # 高重试率 → 检查是否欠分解
        for task in supervisor.tasks:
            if task.retry_count > 3:
                subtasks = decompose_smaller(task)
                supervisor.replace_task(task, subtasks)
                actions_taken.append(f"拆分了任务 {task.id} 为 {len(subtasks)} 个子任务")

    if health_report["escalation_rate"] > 0.2:
        # 高上报率 → 检查 Worker 能力匹配
        for assignment in supervisor.task_assignments:
            if assignment.worker.capability < assignment.task.required_capability:
                new_worker = find_better_worker(assignment.task)
                supervisor.reassign(assignment.task, new_worker)
                actions_taken.append(f"重新分配 {assignment.task.id} 到更强的 Worker")

    return actions_taken

class IntelligentSupervisor:
    """
    自监控 Supervisor：
    1. 任务分解时自动检查分解质量
    2. Worker 失败时自动路由到正确的恢复策略
    3. 持续监控工作流健康度，自动调整分解策略
    """

    def __init__(self, workers: list[Worker], config: dict):
        self.workers = workers
        self.config = config
        self.recovery = RecoveryStrategy()
        self.health_monitor = SupervisorHealthMonitor()
        self.checkpoint_manager = CheckpointManager()

    def decompose_and_validate(self, task: Task) -> list[Task]:
        """分解任务并验证分解质量"""
        subtasks = self._decompose(task)

        # 质量检查
        issues = check_decomposition_quality(subtasks)
        if issues:
            # 尝试自动修复
            subtasks = self._auto_fix_decomposition(subtasks, issues)

        return subtasks

    def execute_with_recovery(self, task: Task, worker: Worker) -> dict:
        """带错误恢复的任务执行"""
        context = {"retry_count": 0, "task_complexity": "normal"}

        while context["retry_count"] < 5:
            result = worker.execute(task, context)

            if result.is_success():
                self.health_monitor.record_task_outcome(task.id, {
                    "status": "success",
                    "duration_seconds": result.duration,
                })
                self.checkpoint_manager.save_checkpoint(task.id, result)
                return result

            # 分类错误
            recovery_action = self.recovery.recover(
                result.error, task, context
            )

            if recovery_action["action"] == "retry":
                context["retry_count"] += 1
                continue

            elif recovery_action["action"] == "escalate":
                self.health_monitor.record_task_outcome(task.id, {
                    "status": "escalated",
                    "duration_seconds": result.duration,
                    "error": result.error,
                })
                return self._escalate_to_human(task, recovery_action)

            elif recovery_action["action"] == "reschedule":
                # 重新拓扑排序后重试
                new_order = recovery_action["new_order"]
                return self._reschedule_and_execute(new_order)

            elif recovery_action["action"] == "refine_task":
                # 拆分任务后重新分配
                refined = self._refine_task(task)
                return self.execute_with_recovery(refined, worker)

        # 达到最大重试次数，上报
        return self._escalate_to_human(task, {"reason": "max_retries_exceeded"})

    def run(self, workflow: Workflow) -> dict:
        """运行完整工作流"""
        # Step 1: 分解 + 验证
        tasks = self.decompose_and_validate(workflow.root_task)

        # Step 2: 拓扑排序，确定执行顺序
        execution_order = topological_sort(tasks)

        # Step 3: 执行 + 监控
        for task in execution_order:
            worker = self._select_worker(task)
            result = self.execute_with_recovery(task, worker)

            if result.is_failed() and result.needs_human():
                return {"status": "needs_review", "task": task, "result": result}

        # Step 4: 健康度检查
        if self.health_monitor.should_self_adjust():
            adjustments = self.self_adjust()
            return {"status": "completed_with_adjustments", "adjustments": adjustments}

        return {"status": "completed", "results": execution_order}