Agent Harness 工程：Agent Team 与多 Agent 协作

上一篇为 Agent Harness 增加了 Scheduler。时间规则到期后，Scheduler 会生成稳定的 ScheduleRun，再创建一个新的 Task。Agent 即使没有收到用户即时输入，也能按照约定时间开始工作。

但当前 Harness 仍然主要依赖一个 Agent 处理任务图。即使前端、后端和测试任务互不依赖，它也只能逐项推进。

第 7 章已经实现过 Subagent。主 Agent 可以把一个局部问题交给干净上下文中的子 Agent，等它返回压缩结果后继续执行。Subagent 更像临时协作者：接受一次委派，完成任务，返回结果，然后退出。

对于持续时间更长、专业分工更明确的大型任务，我们还需要另一种组织方式：

Lead Agent 负责拆解目标、维护任务图和整合结果；
Backend Agent 负责服务端与数据层；
Frontend Agent 负责界面与交互；
QA Agent 负责测试与验证；
Design Agent 负责设计约束与体验检查。

这些 Agent 有稳定身份、独立上下文和不同工具权限。它们共享 Task System，通过 Message Bus 交换必要信息，并在没有工作时休眠，而不是不断调用模型。

本章实现一个最小 Agent Team，解决五个问题：

Team Member 与 Subagent 有什么区别；
如何根据角色和能力把 Task 路由给合适的 Agent；
多个 Agent 如何通信，又不把聊天当成任务状态；
每个 Agent 如何保持上下文隔离并安全交接结果；
Agent 崩溃、重复消息和并发修改时如何恢复。

Agent Team 不是更多 Subagent

最直接的做法似乎是同时调用多次 run_subagent()。它能实现并行，却还不是一个团队。

对比维度	Subagent	Team Member
身份	一次工具调用里的临时实例	稳定 `agent_id` 与长期角色
生命周期	完成一次委派后退出	可以跨多个 Task 休眠和再次被唤醒
任务来源	主 Agent 直接传入任务包	从共享 Task System 认领兼容任务
上下文	一次性干净上下文	每个 Task 使用独立上下文，保留稳定角色
通信	最终结果作为工具结果返回	持久化邮箱，可与 Lead 或队友直接通信
恢复	调用失败后通常整体重试	依赖租约、checkpoint、消息和 artifact
协作关系	主从式、同步等待	多个成员并行推进，通过任务依赖汇合

Team Member 的“长期”不是保留一条无限增长的 messages，也不是让模型一直运行。长期存在的是身份、角色、邮箱和恢复能力；具体执行某个 Task 时，仍然应该创建任务级上下文。

团队的三种共享事实

一个团队至少需要共享三类信息：

Task Store：记录要做什么、依赖、所有权、checkpoint 和结果；
Message Bus：传递请求、进度、阻塞和结果通知；
Artifact Store：保存代码提交、补丁、文档、报告和日志路径。

每个 Agent 保持独立上下文和工具权限，只通过结构化 Task、Message 与 Artifact 交换必要信息。

三者不能互相替代：

Message 说“接口已经完成”不会让 Task 自动变成 completed；
Task 进入 completed 也不应该把完整实现过程复制给所有成员；
Artifact 只是产物引用，必须由 Task 结果说明它解决了什么、如何验证。

最重要的规则是：

Task Store 是工作状态的唯一事实来源，Message Bus 只负责协调和唤醒。

定义稳定的成员档案

Agent Team 不应该让模型临时发明角色、工具和权限。Harness 先注册一组可审计的成员档案：

from dataclasses import dataclass
from typing import Literal


WorkspaceMode = Literal["read_only", "isolated_write", "integration"]


@dataclass(frozen=True)
class AgentProfile:
    id: str
    role: str
    capabilities: frozenset[str]
    tool_names: tuple[str, ...]
    workspace_mode: WorkspaceMode
    role_instructions: str
    max_active_tasks: int = 1


TEAM_PROFILES = {
    "lead": AgentProfile(
        id="lead",
        role="Team Lead",
        capabilities=frozenset({"planning", "review", "integration"}),
        tool_names=(
            "task_create",
            "task_list",
            "task_get",
            "task_claim",
            "task_checkpoint",
            "task_release",
            "task_complete",
            "task_fail",
            "task_retry",
            "run_todo_write",
            "read_file",
            "read_tool_artifact",
            "edit_file",
            "bash",
            "job_status",
            "job_output",
            "job_cancel",
            "job_list",
            "team_list_members",
            "team_start_member",
            "team_send_message",
            "team_inbox",
            "team_task_status",
        ),
        workspace_mode="integration",
        role_instructions="拆解目标、协调依赖、审查结果，不包办所有实现。",
    ),
    "backend": AgentProfile(
        id="backend",
        role="Backend Engineer",
        capabilities=frozenset({"backend", "database", "api"}),
        tool_names=(
            "task_list",
            "task_get",
            "task_claim",
            "task_checkpoint",
            "task_release",
            "task_complete",
            "task_fail",
            "run_todo_write",
            "read_file",
            "read_tool_artifact",
            "edit_file",
            "bash",
            "job_status",
            "job_output",
            "job_cancel",
            "job_list",
            "team_send_message",
            "team_inbox",
        ),
        workspace_mode="isolated_write",
        role_instructions="负责服务端、数据层和 API，并提供可验证的接口契约。",
    ),
    "frontend": AgentProfile(
        id="frontend",
        role="Frontend Engineer",
        capabilities=frozenset({"frontend", "ui", "accessibility"}),
        tool_names=(
            "task_list",
            "task_get",
            "task_claim",
            "task_checkpoint",
            "task_release",
            "task_complete",
            "task_fail",
            "run_todo_write",
            "read_file",
            "read_tool_artifact",
            "edit_file",
            "bash",
            "job_status",
            "job_output",
            "job_cancel",
            "job_list",
            "team_send_message",
            "team_inbox",
        ),
        workspace_mode="isolated_write",
        role_instructions="负责界面、状态与交互，并遵守设计系统和可访问性约束。",
    ),
    "qa": AgentProfile(
        id="qa",
        role="QA Engineer",
        capabilities=frozenset({"testing", "verification"}),
        tool_names=(
            "task_list",
            "task_get",
            "task_claim",
            "task_checkpoint",
            "task_release",
            "task_complete",
            "task_fail",
            "run_todo_write",
            "read_file",
            "read_tool_artifact",
            "bash",
            "job_status",
            "job_output",
            "job_cancel",
            "job_list",
            "team_send_message",
            "team_inbox",
        ),
        workspace_mode="integration",
        role_instructions="验证验收条件、复现缺陷，并提供精简且可定位的测试结果。",
    ),
    "design": AgentProfile(
        id="design",
        role="Product Designer",
        capabilities=frozenset({"design", "ux", "accessibility"}),
        tool_names=(
            "task_list",
            "task_get",
            "task_claim",
            "task_checkpoint",
            "task_release",
            "task_complete",
            "task_fail",
            "run_todo_write",
            "read_file",
            "read_tool_artifact",
            "team_send_message",
            "team_inbox",
        ),
        workspace_mode="read_only",
        role_instructions="审查信息结构、交互与可访问性，并输出明确的设计约束。",
    ),
}

agent_id 是稳定逻辑身份，重启后仍然可以叫 backend；Task 的 claim_token 则属于本次运行实例，每次成员线程或进程启动时重新生成。稳定身份不能代替租约所有权令牌；这个随机 token 也只保护 Task Store 内部写入，并不是外部系统通用的分布式 fencing token。

tool_names 是权限上限，不是提示词建议。Prompt Assembler 只向该成员暴露允许的工具 Schema，工具处理器仍然根据 agent_id、工作区和 Permission Policy 做服务端校验。

长期身份不等于长期上下文

如果 Backend Agent 连续完成十几个 Task，却始终复用同一条对话历史，它最终会遇到和单 Agent 相同的问题：旧日志、旧错误和过期决策不断挤占上下文。

更稳妥的结构是两层状态：

状态层	保存内容
成员级状态	Profile、稳定记忆、邮箱摘要、当前 Task ID
Task 级上下文	当前任务、相关依赖结果、未读消息、必要文件和工具结果

成员认领新 Task 时，Harness 创建新的 Task Session，只注入：

基础 System Prompt；
当前 AgentProfile 的角色和权限；
当前 Task 与少量依赖结果；
与当前 Task 相关的未读消息；
允许读取的 artifact；
当前工具与资源预算。

Task 完成后，完整 messages 可以丢弃或压缩归档。真正需要跨任务保留的经验进入 Memory，工作进度进入 checkpoint，代码与报告进入 Artifact Store。

这延续了 Subagent 的上下文隔离原则，只是把一次性调用升级成了可重复被唤醒的稳定成员。

给 Task 增加能力约束

Task 已经描述了目标和依赖。团队环境还需要回答两个问题：谁有能力执行它，完成后又该怎样把结果交给下游？

可以把路由条件和执行结果定义为独立结构，并由 Task Store 负责将 JSON 还原成统一对象：

from dataclasses import asdict, dataclass, field


@dataclass
class TaskRouting:
    required_capabilities: list[str] = field(default_factory=list)


@dataclass
class TaskResult:
    summary: str
    verification: list[str] = field(default_factory=list)
    risks: list[str] = field(default_factory=list)
    handoff: str = ""


@dataclass
class Task:
    id: str
    title: str
    description: str
    status: TaskStatus = "pending"
    dependencies: list[str] = field(default_factory=list)
    claimed_by: str | None = None
    claim_token: str | None = None
    lease_expires_at: str | None = None
    checkpoint: str | None = None
    artifacts: list[str] = field(default_factory=list)
    routing: TaskRouting = field(default_factory=TaskRouting)
    result: TaskResult | None = None
    error: str | None = None
    created_at: str = ""
    updated_at: str = ""
    schema_version: int = 1

    @classmethod
    def from_dict(cls, data: dict) -> "Task":
        if not isinstance(data, dict):
            raise ValueError("task file must contain a JSON object")
        if data.get("schema_version") != 1:
            raise ValueError("unsupported task schema version")

        payload = dict(data)
        routing_payload = payload.pop("routing", {})
        result_payload = payload.pop("result", None)

        if not isinstance(routing_payload, dict):
            raise ValueError("task routing must be an object")
        routing = TaskRouting(**routing_payload)

        if isinstance(result_payload, str):
            result = TaskResult(summary=result_payload)
        elif isinstance(result_payload, dict):
            result = TaskResult(**result_payload)
        elif result_payload is None:
            result = None
        else:
            raise ValueError("task result must be a string, object, or null")

        task = cls(**payload, routing=routing, result=result)
        task.validate()
        return task

    def validate(self) -> None:
        if self.status not in TASK_STATUSES:
            raise ValueError(f"invalid task status: {self.status}")
        if not isinstance(self.dependencies, list) or not all(
            isinstance(item, str) and item for item in self.dependencies
        ):
            raise ValueError("task dependencies must be non-empty strings")
        if not isinstance(self.routing.required_capabilities, list) or not all(
            isinstance(item, str) and item.strip()
            for item in self.routing.required_capabilities
        ):
            raise ValueError("required capabilities must be non-empty strings")

        if self.result is not None:
            if not isinstance(self.result.summary, str) or not self.result.summary.strip():
                raise ValueError("task result summary is required")
            if not isinstance(self.result.handoff, str):
                raise ValueError("task result handoff must be a string")
            for field_name in ("verification", "risks"):
                values = getattr(self.result, field_name)
                if not isinstance(values, list) or not all(
                    isinstance(item, str) and item.strip() for item in values
                ):
                    raise ValueError(f"task result {field_name} must contain strings")

    def to_dict(self) -> dict:
        return asdict(self)

当任务没有额外路由条件时，required_capabilities=[] 表示任何具备基础 Task 工具的成员都可以执行。字符串形式的 result 则是只有 summary 的简写；Task Store 在读取边界完成归一化，后续能力检查和结果汇总始终只处理 TaskRouting 与 TaskResult。

能力名称必须来自 Team Registry，不能让模型写入不存在的标签。task_complete 同样要先校验 TaskResult 的字段长度、Task 顶层 artifact 路径与验证条目，再在任务锁内写入，不能把一段任意 JSON 直接塞进任务文件。artifact 继续沿用第 13 章的 Task.artifacts，不在 result 中重复保存同一个事实。

例如 Lead 可以创建下面的任务图：

task-api       required: [backend, api]
task-ui        required: [frontend, ui]
task-api-test  required: [testing, verification]  depends_on: [task-api]
task-e2e       required: [testing, verification]  depends_on: [task-api, task-ui]
task-review    required: [review, integration]     depends_on: [task-e2e]

能力检查必须发生在认领锁内：

def ensure_profile_can_claim(task: Task, profile: AgentProfile) -> None:
    required = set(task.routing.required_capabilities)
    missing = required - profile.capabilities
    if missing:
        raise TaskUnavailableError(
            f"agent {profile.id} lacks capabilities: {sorted(missing)}"
        )


def claim_team_task(
    store: TaskStore,
    profiles: dict[str, AgentProfile],
    task_id: str,
    agent_id: str,
    claim_token: str,
    lease_seconds: int = 300,
) -> Task:
    profile = profiles.get(agent_id)
    if profile is None:
        raise TaskUnavailableError(f"unknown team member: {agent_id}")

    with store.locked():
        tasks = {task.id: task for task in store._list_unlocked()}
        task = tasks.get(task_id)
        if task is None:
            raise TaskUnavailableError(f"task does not exist: {task_id}")

        ensure_profile_can_claim(task, profile)
        return claim_task_unlocked(
            store=store,
            tasks=tasks,
            task=task,
            agent_id=agent_id,
            claim_token=claim_token,
            lease_seconds=lease_seconds,
        )

这里把第 13 章的认领逻辑提取为 _unlocked 版本，避免持有 Task Store 锁时再次进入同一把文件锁。依赖检查、租约和 claimed_by 更新仍然在同一个临界区内完成。

模型不能在 task_claim 参数中声明自己的 capabilities。Harness 根据当前 agent_id 从 Registry 注入 Profile，否则任何 Agent 都可以假装具备 review 或 deployment 能力。

推送任务，还是让成员自己认领

团队调度有两种常见方式：

模式	做法	问题
Push	Lead 指定某个 Agent 执行某个 Task	Lead 容易成为瓶颈，成员离线时任务卡住
Pull	空闲成员自己寻找兼容的 ready Task	多个成员可能同时看到同一任务

最小实现可以采用混合方式：

Dispatcher 根据 capabilities 向合适的空闲成员发送“有任务可领”的唤醒事件；
成员读取最新 Task Store，自行选择兼容的 ready Task；
最终所有权仍然通过 task_claim 在存储锁内竞争；
认领失败的成员重新查询，而不是等待模型 API 重试。

Dispatcher 的选择只是一条提示，Task Store 中的 claimed_by + claim_token + lease 才是所有权事实。这样即使两个 Backend Agent 同时被唤醒，也只有一个能取得同一 Task。

第一版最好限制每个成员同时只持有一个 Task。多个 Task 的上下文、租约和后台 Job 会显著增加恢复难度；真正需要并行时，启动更多成员比让一个成员多任务切换更清晰。

Message Bus 不是群聊记录

Task 依赖可以表达“等谁完成”，却无法表达所有协作信息。例如：

Backend Agent 想让 Frontend Agent 提前确认接口字段；
QA Agent 发现验收条件不明确，需要向 Lead 提问；
Frontend Agent 已经生成截图，希望 Design Agent 复核；
某个成员被外部系统阻塞，需要通知协调者。

这些信息适合走 Message Bus，但消息应该短、结构化、可追踪。它不是把所有 Agent 的完整对话拼成一个大群聊。

先定义消息数据：

from dataclasses import asdict, dataclass, field
from typing import Literal


MessageKind = Literal["request", "update", "result", "blocker"]


@dataclass
class TeamMessage:
    id: str
    sender_id: str
    recipient_id: str
    kind: MessageKind
    body: str
    task_id: str | None = None
    reply_to: str | None = None
    artifact_paths: list[str] = field(default_factory=list)
    created_at: str = ""
    acknowledged_at: str | None = None
    schema_version: int = 1

    def to_dict(self) -> dict:
        return asdict(self)

消息按收件人持久化：

.agent/
└── team/
    └── mailboxes/
        ├── lead/
        │   └── message-7f381a0c8d9f4b20.json
        ├── backend/
        ├── frontend/
        ├── qa/
        └── design/

Message Store 继续复用文件锁、路径校验和原子 JSON 写入。正文和 artifact 数量必须有限；测试日志、代码 diff 和截图应该保存为文件，只在消息中传路径和摘要。

发送消息也要幂等

模型工具调用可能在消息写入后、结果返回前中断。重试时如果生成随机 ID，收件人会看到两条相同请求。

可以使用发送者与 tool_call_id 生成稳定消息 ID：

import uuid
from queue import Queue


def team_message_id(sender_id: str, tool_call_id: str) -> str:
    key = f"{sender_id}:{tool_call_id}"
    digest = uuid.uuid5(uuid.NAMESPACE_URL, key).hex[:16]
    return f"message-{digest}"


def message_definition(message: TeamMessage) -> dict:
    return {
        "sender_id": message.sender_id,
        "recipient_id": message.recipient_id,
        "kind": message.kind,
        "body": message.body,
        "task_id": message.task_id,
        "reply_to": message.reply_to,
        "artifact_paths": message.artifact_paths,
    }


def send_team_message(
    bus: MessageBus,
    profiles: dict[str, AgentProfile],
    wake_queues: dict[str, Queue],
    sender_id: str,
    tool_call_id: str,
    recipient_id: str,
    kind: MessageKind,
    body: str,
    task_id: str | None = None,
    reply_to: str | None = None,
    artifact_paths: list[str] | None = None,
) -> TeamMessage:
    if recipient_id not in profiles:
        raise ValueError(f"unknown team member: {recipient_id}")

    normalized_body = body.strip()
    if not normalized_body or len(normalized_body) > 4000:
        raise ValueError("message body must contain 1 to 4000 characters")

    message = TeamMessage(
        id=team_message_id(sender_id, tool_call_id),
        sender_id=sender_id,
        recipient_id=recipient_id,
        kind=kind,
        body=redact_sensitive_text(normalized_body),
        task_id=task_id,
        reply_to=reply_to,
        artifact_paths=list(dict.fromkeys(artifact_paths or []))[:10],
        created_at=utc_now_text(),
    )
    stored = bus.put_if_absent(message)
    wake_queues[recipient_id].put(
        {
            "kind": "team_message_ready",
            "message_id": stored.id,
        }
    )
    return stored

sender_id 和 tool_call_id 都由 Harness 注入，不接受模型伪造。put_if_absent() 必须在 Message Store 锁内完成“检查并创建”；遇到相同 ID 时，用 message_definition() 比较业务字段并返回原消息，内容冲突则报错。created_at 和 acknowledged_at 不参与定义比较，否则一次合法重试会因为时间戳不同而被误判成冲突。

写入磁盘发生在唤醒之前。即使进程在两步之间退出，收件人启动时扫描未确认消息，也能重新发现它。Harness 初始化 Team Runtime 时，要为所有已注册 Profile 预先创建 wake queue，而不是等成员线程启动后才创建。

消息如何进入 Agent 上下文

Message 不是用户输入，不能追加成普通 user 消息。Prompt Assembler 应该把未确认消息渲染为带来源的运行时 Section：

Team inbox:
- id: message-7f381a0c8d9f4b20
  from: backend
  kind: update
  task: task-api
  body: API implementation is ready. The response contract is attached.
  artifacts: [.agent/artifacts/task-api/openapi.json]

一次模型调用成功后，Runtime 再把本轮注入的消息标记为 acknowledged_at。如果模型调用失败，消息保持未确认，下一轮继续投递。这是跨 Harness 重启的至少一次语义。

至少一次意味着消息可能重复出现。Agent 不应该依靠“我记得已经看过”去重；Runtime 使用稳定 message_id 过滤当前上下文，工具操作则继续使用各自的幂等键。

消息到达时，如果成员正在执行 Task，不要并发启动第二个模型回合去修改同一份状态。唤醒事件先进入该成员的队列，run_task_session() 在当前模型或工具步骤结束后读取并注入新消息。

消息不会解锁下游 Task

假设 Backend Agent 完成接口后，需要 QA Agent 开始测试。

实线表示持久化工作状态：只有上游 Task 完成才会解锁依赖。虚线消息用于提前沟通、发送摘要和唤醒成员。

正确的交接顺序是：

Backend Agent 保存 checkpoint 和 artifact；
Backend Agent 调用 task_complete 写入结构化结果；
task-api-test 因依赖满足变为 ready；
Backend Agent 向 QA Agent 发送简短结果通知；
Dispatcher 或消息事件唤醒 QA Agent；
QA Agent 重新读取 Task Store 并认领测试 Task。

如果消息先到、Backend Task 仍是 in_progress，QA 可以阅读接口说明或提出问题，但不能把依赖尚未满足的测试 Task 强行认领。

同样，消息丢失或延迟也不会永久阻塞工作。Task 进入 completed 后，下游 Task 已经可以被 Dispatcher 发现，消息只是降低协作延迟。

结构化结果比聊天总结更可靠

Team Member 完成 Task 时，结果应该使用前面 TaskResult 的固定字段：

{
  "result": {
    "summary": "实现用户搜索 API，并补充分页与空查询处理。",
    "verification": ["pytest tests/api/test_search.py -q: passed"],
    "risks": ["模糊搜索仍依赖数据库 collation"],
    "handoff": "QA 应重点验证分页边界和非 ASCII 查询。"
  },
  "artifacts": ["commit:8cb7f31", ".agent/artifacts/task-api/openapi.json"]
}

Lead 和下游成员读取的是 Task 结果与 artifact，不需要回放 Backend Agent 的完整对话。task_complete 在锁内写入这份结构化结果后，Message Bus 中的 result 消息只保留更短摘要和 Task ID；消息发送失败也不会回滚已经完成的 Task。

结构化结果也方便恢复：如果 Lead 崩溃，新 Lead 可以从 Task Store 重建当前进度，而不必依赖某个内存聊天窗口。

启动成员与事件驱动循环

Lead 可以唤醒预先注册的 Team Member，但不能在运行时任意注入新的 System Prompt 或工具权限：

from threading import Thread


def ensure_team_member_running(state: HarnessState, agent_id: str) -> None:
    profile = state.team_profiles[agent_id]

    with state.team_lock:
        existing = state.member_threads.get(agent_id)
        if existing and existing.is_alive():
            return

        thread = Thread(
            target=supervised_team_member_loop,
            args=(state, profile),
            name=f"agent-team-{agent_id}",
            daemon=False,
        )
        state.member_threads[agent_id] = thread
        thread.start()

成员循环只在 Task 或 Message 到达时工作：

import secrets
import time
from queue import Empty


def team_member_loop(state: HarnessState, profile: AgentProfile) -> None:
    claim_token = secrets.token_urlsafe(16)
    wake_queue = state.member_wake_queues[profile.id]
    wake_queue.put({"kind": "member_started"})

    while state.accepting_team_work:
        try:
            wake_queue.get(timeout=30.0)
        except Empty:
            # 周期性 reconciliation 可以恢复“事实已落盘、唤醒事件却丢失”的窗口。
            pass

        unread = state.message_bus.list_unacknowledged(profile.id)
        priority_messages = [
            message
            for message in unread
            if message.kind in {"request", "blocker"}
        ]
        if priority_messages:
            run_coordination_turn(
                state=state,
                profile=profile,
                messages=priority_messages,
            )

        # 协调回合可能确认了部分消息；重新读取，不能继续使用旧快照。
        unread = state.message_bus.list_unacknowledged(profile.id)
        if any(message.kind in {"request", "blocker"} for message in unread):
            wake_queue.put({"kind": "team_message_ready"})
            continue

        task = claim_next_compatible_task(
            task_store=state.task_store,
            profiles=state.team_profiles,
            agent_id=profile.id,
            claim_token=claim_token,
        )

        if task is not None:
            task_messages = [
                message
                for message in unread
                if message.task_id in {None, task.id}
            ]
            run_task_session(
                state=state,
                profile=profile,
                task=task,
                claim_token=claim_token,
                initial_messages=task_messages,
            )
            if state.message_bus.list_unacknowledged(profile.id):
                wake_queue.put({"kind": "team_message_ready"})
        elif unread:
            run_coordination_turn(
                state=state,
                profile=profile,
                messages=unread,
            )

        if state.message_bus.list_unacknowledged(profile.id):
            wake_queue.put({"kind": "team_message_ready"})


def supervised_team_member_loop(
    state: HarnessState,
    profile: AgentProfile,
) -> None:
    restart_count = 0

    while state.accepting_team_work:
        try:
            team_member_loop(state, profile)
            return
        except Exception as exc:
            restart_count += 1
            state.record_member_failure(profile.id, exc, restart_count)
            if not state.accepting_team_work:
                return
            if restart_count >= 5:
                state.mark_member_unhealthy(profile.id, "restart budget exhausted")
                return
            time.sleep(min(30, 2 ** min(restart_count - 1, 4)))

空闲成员主要阻塞在本地 Queue 上，不消耗 token。每 30 秒进行一次不调用模型的 reconciliation，是为了覆盖“Task 或 Message 已经持久化，但进程在发出唤醒前崩溃”的窗口；成员启动和 Supervisor 重启时也会立即自唤醒一次。claim_next_compatible_task() 先过滤 capabilities，再逐个尝试原子认领；列表结果只是快照，认领冲突仍然按正常调度竞争处理。

这里没有在成功处理完全部高优先级消息后直接 continue。一次 Task 唤醒和一条 blocker 可能同时到达；如果协调回合消费了 blocker，也必须在同一次唤醒中继续查询 Task Store，否则原来的 Task 唤醒已经出队，成员可能在明明有 ready Task 时重新休眠。若仍有未确认的 request 或 blocker，则重新入队并暂缓认领。任何唤醒都只是“状态可能变化”的提示，成员醒来后统一重查邮箱和任务图。

run_task_session() 每完成一次模型或工具步骤，都应该检查新消息和 Task 租约：注入未读消息、保存 checkpoint、续租，并在退出前调用 task_complete、task_fail 或 task_release。

claim_next_compatible_task() 还要在认领锁内统计该成员仍持有有效租约的 Task，执行 profile.max_active_tasks；Supervisor 重启后即使得到新 token，也不能绕过旧实例尚未过期的任务配额。

成员线程由 Supervisor 包装。模型调用、工具或消息解析抛出异常时，Supervisor 记录成员故障并采用有限退避；同一 Supervisor 生命周期累计 5 次失败后将成员标记为 unhealthy，不再形成永久重启循环。未确认消息继续留在邮箱，Task 则通过 checkpoint 和租约恢复。新循环会生成新的 claim_token，因此它不能继续写旧实例仍在持有的 Task，也不能在没有检查副作用的情况下立即重复执行。

这里已经能启动队友，却还没有解决“如何安全停止队友”。代码中的 accepting_team_work 只是运行开关，不能直接替代正式关机协议，这会留到下一章。

Lead 负责协调，不负责转发一切

Lead Agent 的职责包括：

理解用户目标并创建有依赖的 Task DAG；
为 Task 标注所需 capabilities 和验收条件；
启动需要的 Team Member；
关注 blocked、failed、租约过期和无人可执行的 Task；
审查结构化结果并执行最终集成；
向用户汇总进度、风险和最终结果。

Lead 不应该：

手工覆盖 claimed_by 把任务塞给某个成员；
充当所有成员消息的中转站；
把完整用户对话广播给整个团队；
在成员完成专业工作后无条件重做一遍；
因为自己是 Lead 就绕过工具和权限策略。

Backend 与 Frontend 可以直接确认接口，QA 可以直接向责任成员反馈复现步骤。只有跨任务优先级、需求冲突和最终集成需要回到 Lead。

如果所有信息都必须经过 Lead，团队虽然有多个 Agent，吞吐量仍然会被单个上下文和模型回合限制。

多 Agent 最容易在文件系统里打架

Task 认领只能防止两个 Agent 同时执行同一个 Task，不能阻止两个不同 Task 修改同一个文件。

例如 Backend Agent 修改共享类型，Frontend Agent 同时修改同一文件，即使两个 Task 都合法，也可能互相覆盖。

可选策略包括：

策略	优点	局限
共享工作区	实现最简单	修改容易覆盖，状态难以归属
文件级预留	能提前发现部分冲突	很难预测间接修改和生成文件
每成员工作区	成员环境稳定，便于持续开发	同一成员的多个 Task 仍会串扰
每 Task 工作区	隔离最清晰，artifact 易追踪	创建和集成成本更高

对于可以写代码的 Team Member，推荐使用独立 Git worktree、分支或沙箱。成员完成后返回 commit、patch 和验证结果，由 Integration Workspace 负责合并与最终测试。

测试成员应该验证已经集成的结果，而不是只测试 Backend Agent 的私有工作区。Task 依赖也应该反映集成顺序，例如 task-e2e 依赖前后端合并任务，而不只是依赖两个实现任务。

工作区隔离不能替代权限边界。每个 Profile 仍然只能访问分配给它的路径、环境变量和工具。

成员崩溃后如何恢复

前几章已经准备了大部分恢复能力：

Task 租约：成员崩溃后，Task 到期变为 reclaimable；
claim_token：旧实例恢复后不能继续修改已经被接管的 Task；
checkpoint：新成员知道上一个执行者做到哪里；
Artifact：代码、报告和日志不会只留在对话里；
持久化邮箱：未确认消息在 Harness 重启后仍然存在；
稳定 message ID：重复投递不会产生两份不同请求；
后台 Job 元数据：长操作可以检查结果，而不是盲目重跑。

恢复时不要执着于原角色。如果 Backend Agent 长期离线，只要另一个成员具备所需 capabilities，就可以在租约过期后接手 Task。agent_id 用于审计和协作，不应该成为永久锁死工作的唯一条件。

如果没有任何 Profile 满足 Task 的 required capabilities，Dispatcher 应把它标记为“无人可执行”的派生视图并通知 Lead，而不是持续唤醒所有成员。

暴露 Team 工具

最小团队只需要少量专用工具：

工具	作用
`team_list_members`	查看成员角色、能力和当前运行状态
`team_start_member`	启动已注册但尚未运行的成员
`team_send_message`	向指定成员发送结构化消息
`team_inbox`	按任务或发送者查看未确认与历史消息
`team_task_status`	汇总团队正在处理、阻塞和可认领的 Task

team_send_message 的 Tool Schema 不暴露 sender_id：

TEAM_SEND_MESSAGE_TOOL = {
    "type": "function",
    "function": {
        "name": "team_send_message",
        "description": "Send a concise coordination message to a team member.",
        "parameters": {
            "type": "object",
            "properties": {
                "recipient_id": {
                    "type": "string",
                    "enum": sorted(TEAM_PROFILES),
                },
                "kind": {
                    "type": "string",
                    "enum": ["request", "update", "result", "blocker"],
                },
                "task_id": {"type": "string"},
                "reply_to": {"type": "string"},
                "body": {
                    "type": "string",
                    "maxLength": 4000,
                },
                "artifact_paths": {
                    "type": "array",
                    "items": {"type": "string"},
                    "maxItems": 10,
                },
            },
            "required": ["recipient_id", "kind", "body"],
        },
    },
}

消息工具也要经过权限和审计：发送者是否能查看关联 Task、artifact 是否位于允许路径、收件人是否有权接收这类信息，都由 Harness 校验。

Message 不会授予新权限。Lead 发一句“请执行部署”，不能让没有 deployment capability 或网络权限的成员获得部署能力。

预算与并发边界

多个 Agent 会同时放大模型调用、工具调用和外部副作用。Harness 至少需要限制：

团队总成员数和每个角色的实例数；
同时运行的模型回合数量；
每个成员和每个 Task 的 token、时间与费用预算；
每个成员持有的 Task 数量；
消息频率、正文长度和未确认消息数量；
后台 Job 总并发数；
写工作区、网络与高风险工具的权限；
Lead 创建新 Task 和启动新成员的速率。

不要让 Agent 因为收到每一条 update 都立刻调用模型。Message Bus 可以合并同一 Task 的普通进度通知；只有 request、blocker 或真正解锁工作的事件需要及时唤醒。

团队扩容也不总会加快任务。任务依赖、共享文件、模型限流和集成成本都可能让更多成员降低整体效率。Dispatcher 应根据 ready Task 数量和能力缺口启动成员，而不是固定把所有角色全部拉起。

常见坑

第一，把 Agent Team 实现成并发 run_subagent()。 这缺少稳定身份、共享任务所有权、邮箱和恢复能力。

第二，让所有成员共享一份 messages。 上下文会互相污染，也无法判断某个决策属于谁。

第三，把长期成员理解成永不清空上下文。 稳定身份应该跨 Task，Task Session 不应该无限增长。

第四，让 Lead 直接修改 claimed_by。 所有权必须通过带锁的 task_claim 和租约建立。

第五，只靠 Message 宣布任务完成。 下游是否解锁必须读取 Task Store。

第六，把完整日志和代码贴进 Message。 大内容保存为 artifact，消息只传摘要和路径。

第七，信任模型声明自己的角色和 capabilities。 身份、能力和工具权限必须来自 Harness Registry。

第八，多个写 Agent 共用同一工作区。 不同 Task 也可能修改相同文件，需要 worktree、分支或明确集成流程。

第九，每条进度消息都唤醒模型。 这会制造消息风暴和不可控费用。

第十，让 Lead 转发所有沟通。 Lead 会成为吞吐瓶颈，队友应该围绕 Task 直接协作。

第十一，成员崩溃后立刻重复执行。 先等待租约、读取 checkpoint，并检查已有 artifact 与后台 Job。

第十二，用强制终止作为正常停机方式。 正在写文件或持有 Task 的成员可能留下不完整状态，这正是下一章要解决的问题。

小结

本文在 Subagent、Task System 和 Scheduler 之上，实现了一个最小 Agent Team：

区分一次性 Subagent 与具有稳定身份、角色和邮箱的 Team Member；
让长期成员为每个 Task 使用独立上下文，避免历史无限膨胀；
通过 AgentProfile 固定 capabilities、工具权限和工作区模式；
为 Task 增加 required capabilities，并在认领锁内校验；
用 Dispatcher 唤醒合适成员，最终仍由 Task Store 原子认领决定所有权；
用持久化 Message Bus 传递请求、进度、结果和阻塞信息；
使用稳定 message ID 与确认机制提供跨重启的至少一次投递；
保持 Task Store 为工作状态事实，Message 只做协调和唤醒；
通过结构化结果、artifact 和隔离工作区完成安全交接；
复用租约、claim_token、checkpoint 和后台 Job 支持成员故障恢复；
对团队规模、模型并发、消息频率、工作区和费用设置明确边界。

至此，Agent 已经不再只是临时调用几个子代理，而是能够像一个小型开发团队一样，让不同角色长期协作、并行认领任务，并把结果汇总回同一张任务图。

但团队能启动、能工作、能通信，并不代表它能安全退出。如果 Lead 直接杀掉 backend 线程，成员可能正在写文件、等待后台 Job，或者持有尚未保存 checkpoint 的 Task。

下一章讲 Agent Team 协议与优雅退出：由 Lead 发送 shutdown_request，成员停止认领新任务，保存 checkpoint、处理正在运行的 Job，并在完成交接后返回 shutdown_ack；Harness 再根据超时和风险决定是否升级为强制终止。