协作终止与故障传播

协作终止是判断整个团队何时停止，而不是只看某位Agent是否说“完成”。故障传播决定一个子任务失败后，哪些后续任务不能继续，哪些仍可产出有用结果。

查退款资格必须拿到订单和规则。润色文字可以让答复更顺，但它不是判断资格所必需的资料。订单读取失败和润色失败，不应该产生相同的团队状态。

集中终止由协调者统一检查完成条件、关键失败和预算。如果每位Agent都能单独宣布全队成功，协作就会变成“我下班了，所以公司今天任务全部完成”。

订单成功、规则失败时，队伍怎样停

Team跟踪order、policy和style三项任务，REQUIRED指定order与policy必须完成。初始状态pending表示待处理，working表示正在处理，completed与failed表示子任务已结束。

订单先完成，团队仍为running。规则随后失败，它是必需项，团队进入critical_failure，尚未完成的style变为canceled。之后即使style送来completed，团队也返回late_ignored，不再改写已停止状态。

另两组固定事件展示步数耗尽与可选任务失败。每处理一个有效事件消耗一步；达到上限且必需任务尚未齐全，就进入step_limit。style失败后，只要订单和规则都完成，团队仍可completed。

保存完整程序并运行

在自己的学习目录打开PowerShell，创建一个新的子目录：

mkdir l6-termination
cd l6-termination

把下面完整代码保存为 termination_demo.py。固定数据和运行入口都已包含，文件名保留 .py 后缀。

"""集中管理团队状态，传播关键故障并忽略迟到结果。"""
REQUIRED = {"order", "policy"}

class Team:
    def __init__(self, max_steps=4):
        if type(max_steps) is not int or max_steps < 1:
            raise ValueError("步数上限必须是正整数")
        self.max_steps = max_steps
        self.steps = 0
        self.status = "running"
        self.tasks = {"order": "pending", "policy": "pending", "style": "pending"}

    def stop(self, reason):
        self.status = reason
        for name, state in self.tasks.items():
            if state in {"pending", "working"}:
                self.tasks[name] = "canceled"

    def apply(self, name, state):
        if self.status != "running":
            return "late_ignored"
        if name not in self.tasks or state not in {"working", "completed", "failed"}:
            raise ValueError("未知子任务或状态")
        if self.tasks[name] in {"completed", "failed", "canceled"}:
            raise ValueError("子任务已结束")
        self.steps += 1
        self.tasks[name] = state
        # 同一步里关键失败优先；正常完成优先于耗尽步数。
        if name in REQUIRED and state == "failed":
            self.stop("critical_failure")
        elif all(self.tasks[required] == "completed" for required in REQUIRED):
            self.stop("completed")
        elif self.steps >= self.max_steps:
            self.stop("step_limit")
        return self.status

def main():
    team = Team()
    print("订单读取：" + team.apply("order", "completed"))
    print("规则读取：" + team.apply("policy", "failed"))
    print("其他任务：" + ",".join(f"{name}={state}" for name, state in team.tasks.items()))
    print("迟到结果：" + team.apply("style", "completed"))
    small = Team(max_steps=2)
    small.apply("order", "working")
    print("重复等待：" + small.apply("policy", "working"))
    optional = Team()
    optional.apply("style", "failed")
    optional.apply("order", "completed")
    print("可选任务失败后：" + optional.apply("policy", "completed"))

if __name__ == "__main__":
    main()

保存后执行 dir，确认能看到 termination_demo.py，再运行：

ls
python -X utf8 termination_demo.py

程序输出：

订单读取：running
规则读取：critical_failure
其他任务：order=completed,policy=failed,style=canceled
迟到结果：late_ignored
重复等待：step_limit
可选任务失败后：completed

为什么先检查关键失败，再看步数

apply先核对任务名和状态，非法事件不消耗额度。有效事件到达后增加steps、更新该任务状态，然后按顺序判断关键失败、必需任务全完成和步数上限。

这个顺序解决同一步触发多个条件的问题。第二步正好完成全部必需任务，同时也到达max_steps=2，结果是completed，而不是误报step_limit。关键任务失败则保持critical_failure，不被耗尽条件遮住真正原因。

stop把其他pending或working状态设为canceled。已经完成的order保留completed，失败的policy保留failed，只有未完成的style被取消。保留这些子状态，方便知道团队为什么停、哪些资料已经拿到。

团队状态与子任务状态不同

团队completed表示这份任务定义中的必需项完成。它不要求所有可选项都成功。例子允许style失败，因为候选结论的依据完整；若用户要求必须得到润色后的报告，style就应该进入REQUIRED。

团队停止后，late_ignored表示本地协调者忽略迟到结果，不重新开启团队。接收器仍应记录来源和到达时间，方便对账；忽略业务写入不等于删除故障证据。

这里canceled是协调者的控制状态，程序没有启动并发工作进程。真实队伍需要向运行中的执行器发取消信号，并等待其确认；网络请求、工具调用和写入动作能否取消，要看执行系统的能力。

尤其是付款等副作用，标记canceled不能把已执行的动作撤回。终止记录应保留哪些动作已经开始、哪些已经确认、哪些结果仍未知，之后按幂等键和对账流程处理迟到结果。

只有步数上限还不够

max_steps限制有效事件数量，不能限制一条事件等待了多久。实际任务还需要总截止时间、单次调用超时、输出上限和费用额度；其中任意一项到达后，要按已定义优先级停止或转人工。

心跳或重复“正在查”也不等于业务进展。若一直收到working事件，却没有新资料或已完成子任务，应设置无进展上限。无进展可以基于新增产物、未决任务数量或明确状态变化判断，而不是读取模型的隐藏思考。

关键子任务的错误可细分为短暂网络失败、资料缺失和权限拒绝。短暂失败可在有限预算内重试；资料缺失可能等用户补充；权限拒绝不应靠原样重试绕过。故障类型应帮助选择后续动作，而不是只产生一句“失败了”。

把可选任务改成必需任务

把REQUIRED改为{"order", "policy", "style"}，运行python -X utf8 termination_demo.py。最后一组事件中style失败会立刻造成critical_failure，随后订单与规则事件都变成迟到事件，最终不会显示completed。

恢复REQUIRED，再把第二组Team(max_steps=2)的2改为3，运行同一命令。两个working事件之后仍为running，说明额度没有用尽；这不代表任务成功，只说明还可以处理后续事件。

将这处max_steps恢复为2，确认REQUIRED仍只有order和policy，再运行一次。重复等待回到step_limit，可选任务失败后的团队仍为completed，接着运行配套检查。

看到step_limit时，核对steps与未完成必需项；看到critical_failure时，定位失败子任务与原因。增加轮数可以延后停止，却不能补出一直失败的订单证据。

参考阅读：AutoGen Termination比较消息数量、超时与组合停止条件；A2A任务生命周期关注中断和终态如何影响后续交互。

运行配套检查

把下面测试完整保存为 test_termination_demo.py，与主程序放在同一目录。6项检查覆盖关键失败传播、迟到结果、可选失败、最后一步完成、额度耗尽和非法事件。

"""测试关键失败、可选失败、同一步优先级和迟到消息。"""
import sys
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent))
import termination_demo as demo

class TerminationTests(unittest.TestCase):
    def test_required_failure_cancels_unfinished_work(self):
        team = demo.Team()
        self.assertEqual(team.apply("policy", "failed"), "critical_failure")
        self.assertEqual(team.tasks["order"], "canceled")
        self.assertEqual(team.tasks["style"], "canceled")

    def test_late_result_cannot_reopen_team(self):
        team = demo.Team()
        team.apply("policy", "failed")
        before = dict(team.tasks)
        self.assertEqual(team.apply("order", "completed"), "late_ignored")
        self.assertEqual(team.tasks, before)

    def test_optional_failure_does_not_fail_required_work(self):
        team = demo.Team()
        team.apply("style", "failed")
        team.apply("order", "completed")
        self.assertEqual(team.apply("policy", "completed"), "completed")
        self.assertEqual(team.tasks["style"], "failed")

    def test_completion_on_last_step_wins_over_limit(self):
        team = demo.Team(max_steps=2)
        team.apply("order", "completed")
        self.assertEqual(team.apply("policy", "completed"), "completed")

    def test_incomplete_work_hits_limit(self):
        team = demo.Team(max_steps=2)
        team.apply("order", "working")
        self.assertEqual(team.apply("policy", "working"), "step_limit")

    def test_invalid_event_does_not_consume_step(self):
        team = demo.Team()
        with self.assertRaises(ValueError):
            team.apply("unknown", "completed")
        self.assertEqual(team.steps, 0)

if __name__ == "__main__":
    unittest.main()

python -X utf8 -m unittest -v test_termination_demo.py

测试名称后应显示 ok，末尾显示 Ran 6 tests 和 OK。若找不到模块，先用 dir 确认两个文件保存在同一目录。