## 条件、并行与循环

控制节点改变的是 token 的路由和作用域，不是普通动作节点上的隐藏条件。
可靠设计必须明确哪些分支会被激活、何时汇合、失败怎样传播，以及循环中的一次节点访问属于哪次迭代。

<!-- slot: 目标 -->
### 本章目标

学完本章，你将能够：

- 使用 `switch` 表达确定性的单路或多路条件选择。
- 使用 `fork` 和配对 `join` 建立固定并行作用域。
- 区分 `join.mode: all` 与 `join.mode: any` 的成功和失败语义。
- 使用 `loop.mode: counted` 表达精确次数的循环与可选提前成功退出。
- 避免并行状态竞争、跨迭代串线和无法关闭的图作用域。

<!-- slot: 边界 -->
### 适用边界

- **覆盖**：`switch`、`fork`、固定并行 `join` 与计数 `loop`。
- **不覆盖**：由 Agent 返回 WorkItem 后动态展开的 `fan_out`，见第 14 章。
- **假设**：分支内使用的动作节点遵守第 11 章的 Effect、重试和幂等边界。

### 条件路由必须显式

`switch.mode: first_match` 按作者顺序求值，只发送第一个匹配端口。
`switch.mode: all_match` 求值全部条件，并并行发送所有匹配端口。
每个 `switch` 恰好有一个默认 case；`first_match` 在此前没有匹配时选择它，`all_match` 仅在所有条件 case 都不匹配时选择它。

条件是无副作用且有类型的表达式。
普通动作边没有隐藏 `when`；需要条件时就使用 `switch`。
`all_match` 形成条件并行 lineage，必须由命名该 `switch` 的 `join.forkId` 关闭；`join` 只等待本次实际激活的分支集合。

下面的最小 Plan 把高风险值路由到成功终点，把未声明值路由到稳定失败终点。

```yaml
apiVersion: loop.plan/v1
kind: PlanVersion
input:
  severity:
    type: string
    required: true
graph:
  entry:
    nodeId: route
    port: in
    input:
      severity: {ref: plan.input.severity}
  nodes:
    - id: route
      type: switch
      input:
        severity:
          type: string
          required: true
      switch:
        mode: first_match
        cases:
          - port: escalate
            when: 'input.severity == "critical" || input.severity == "high"'
          - port: ordinary
            when: 'input.severity == "normal"'
          - port: invalid
            isDefault: true
    - id: success
      type: end
      end: {outcome: succeeded}
    - id: failure
      type: end
      end: {outcome: failed, code: severity_unhandled}
  edges:
    - id: escalated
      source: {nodeId: route, port: escalate}
      target: {nodeId: success, port: in}
    - id: ordinary
      source: {nodeId: route, port: ordinary}
      target: {nodeId: success, port: in}
    - id: invalid
      source: {nodeId: route, port: invalid}
      target: {nodeId: failure, port: in}
```

### 固定并行由 lineage 关联

`fork` 无条件激活所有已声明分支。
它与 `switch: all_match` 的区别在于前者没有条件选择。
配对 `join` 使用 Run、join 节点和 fork 激活身份关联到达值，因此嵌套并行和循环迭代不会混收 token。

`join.mode: all` 等待本次激活的每个必需分支各有一次成功到达。
如果某个必需分支不能成功，它按显式失败路由结束。
`join.mode: any` 接受第一个**成功**到达，而不是第一个终态；它原子关闭 join，并取消同一 fork 激活中的剩余分支。
取消会阻止新的图工作并使迟到 callback 失效，但不能撤回已对外可见的消息。

`all` 模式从 `branches.<分支键>.*` 显式合并结果。
`any` 模式从 `winner.branch` 和 `winner.value.*` 读取胜出分支。
runtime 不按 callback 到达顺序自动浅合并、深合并或采用最后写入值。

<!-- slot: 提示词 -->
### 提示词

本章不依赖自然语言创建。
评审并行图时，可以使用下面的检查清单：

```text
列出每个 switch 或 fork 可能激活的分支集合。
为每个集合指出唯一配对 join、关闭条件和失败传播。
检查任意两条可并行分支是否读写同一个 Run 状态。
检查 join 后的结果是否按稳定分支键显式合并。
```

<!-- slot: 流程 -->
### 固定并行流程

下面的 Plan 无条件复制两个 token，并在 `gather` 汇合：

1. `fanout` 为 `left` 和 `right` 各创建一个带 lineage 的分支。
2. 两条边把同一个只读值送到不同 join 端口。
3. `gather` 等待全部分支成功到达。
4. `merge` 明确选择 `branches.left.value` 作为 Plan 输出。
5. join 失败进入 Plan 的失败终点 `join_failed`，不会假装成功。

<!-- slot: 配置 -->
### 固定并行配置

```yaml
apiVersion: loop.plan/v1
kind: PlanVersion
input:
  value: {type: string, required: true}
output:
  value: {type: string, required: true}
graph:
  entry:
    nodeId: fanout
    port: in
    input:
      value: {ref: plan.input.value}
  nodes:
    - id: fanout
      type: fork
      input:
        value: {type: string, required: true}
      fork:
        branches: [left, right]
    - id: gather
      type: join
      input:
        value: {type: string, required: true}
      join:
        forkId: fanout
        mode: all
        result:
          value: {type: string, required: true}
        merge:
          value: {ref: branches.left.value}
    - id: success
      type: end
      end: {outcome: succeeded}
    - id: failure
      type: end
      end: {outcome: failed, code: join_failed}
  edges:
    - id: left
      source: {nodeId: fanout, port: left}
      target: {nodeId: gather, port: left}
      bindings:
        value: {ref: input.value}
    - id: right
      source: {nodeId: fanout, port: right}
      target: {nodeId: gather, port: right}
      bindings:
        value: {ref: input.value}
    - id: joined
      source: {nodeId: gather, port: completed}
      target: {nodeId: success, port: in}
      bindings:
        value: {ref: output.value}
    - id: join_failed
      source: {nodeId: gather, port: failed}
      target: {nodeId: failure, port: in}
```

这份配置来自 [`internal/workflow/plan/compiler_test.go: validForkJoinPlan`](https://github.com/tidbcloud/AgentTeam/src/branch/main/internal/workflow/plan/compiler_test.go) 的固定并行 fixture。
编译器在 [`internal/workflow/plan/parallel.go: validator.validateParallelScopes`](https://github.com/tidbcloud/AgentTeam/src/branch/main/internal/workflow/plan/parallel.go) 中证明 fork 对 join 的支配关系、分支成员互斥、每条分支的成功边界和 Run 状态访问安全。
runtime 的 `all` 与 `any` 行为分别由 [`internal/workflow/run/reducer_test.go: TestForkJoinAllAndScopeTransitionsAsSet`](https://github.com/tidbcloud/AgentTeam/src/branch/main/internal/workflow/run/reducer_test.go) 和 [`internal/workflow/run/reducer_test.go: TestJoinAnyChoosesFirstSuccessfulNotFirstTerminal`](https://github.com/tidbcloud/AgentTeam/src/branch/main/internal/workflow/run/reducer_test.go) 覆盖。

### 计数循环不是安全上限

`loop.plan/v1` 只支持 `mode: counted`。
`iterations` 必须为正数，表示**恰好执行多少次 body traversal**，而不是“最多尝试多少次”。
进入循环会发出第 1 次 `body`；每次 `continue` 原子替换循环状态。
第 N 次 `continue` 自动成功并从 `done` 端口离开。
可选 `break` 是提前成功退出，并使用与最终结果相同的状态契约。

进入 `loop` 时 runtime 只创建一个 loop scope；后续 `continue` 复用该 scope，替换其状态并递增 `LoopState.Iteration`。
每次访问静态节点仍会创建新的 Activation；循环体每次执行 `fork` 时也会新建 parallel scope 和 fork Activation，因此不同迭代的 join 不会互相消费 token。
循环外的任意回边、自环或不经 `continue` 形成的周期都会被编译器拒绝。

```yaml
apiVersion: loop.plan/v1
kind: PlanVersion
input:
  answer: {type: string, required: true}
output:
  answer: {type: string, required: true}
graph:
  entry:
    nodeId: repeat
    port: enter
    input:
      answer: {ref: plan.input.answer}
  nodes:
    - id: repeat
      type: loop
      loop:
        mode: counted
        iterations: 3
        state:
          answer: {type: string, required: true}
    - id: route
      type: switch
      input:
        answer: {type: string, required: true}
      switch:
        mode: first_match
        cases:
          - port: again
            when: 'input.answer == "again"'
          - port: stop
            when: 'contains(input.answer, "stop") == true'
          - port: fallback
            isDefault: true
    - id: success
      type: end
      end: {outcome: succeeded}
    - id: failure
      type: end
      end: {outcome: failed, code: loop_failed}
  edges:
    - id: body
      source: {nodeId: repeat, port: body}
      target: {nodeId: route, port: in}
      bindings:
        answer: {ref: loop.state.answer}
    - id: again
      source: {nodeId: route, port: again}
      target: {nodeId: repeat, port: continue}
      bindings:
        answer: {ref: input.answer}
    - id: stop
      source: {nodeId: route, port: stop}
      target: {nodeId: repeat, port: break}
      bindings:
        answer: {ref: input.answer}
    - id: fallback
      source: {nodeId: route, port: fallback}
      target: {nodeId: repeat, port: break}
      bindings:
        answer: {ref: input.answer}
    - id: done
      source: {nodeId: repeat, port: done}
      target: {nodeId: success, port: in}
      bindings:
        answer: {ref: loop.state.answer}
    - id: loop_failed
      source: {nodeId: repeat, port: failed}
      target: {nodeId: failure, port: in}
```

编译器的精确次数与状态投影由 [`internal/workflow/plan/counted_loop_test.go: TestCompileCountedLoopPublishesExactCountAndFinalStateProjection`](https://github.com/tidbcloud/AgentTeam/src/branch/main/internal/workflow/plan/counted_agentteam_test.go) 验证，作用域闭合规则见 [`internal/workflow/plan/loop_scope_test.go: TestCompileRejectsCrossLoopBoundary / TestCompileRejectsLoopBodyEscape`](https://github.com/tidbcloud/AgentTeam/src/branch/main/internal/workflow/plan/agentteam_scope_test.go)。
runtime 的精确 body 次数、提前退出和持久状态 JSON 往返分别由 [`internal/workflow/run/counted_loop_test.go: TestCountedLoopRunsExactlyNBodyTraversalsThenAutomaticallySucceeds`](https://github.com/tidbcloud/AgentTeam/src/branch/main/internal/workflow/run/counted_agentteam_test.go)、[`internal/workflow/run/counted_loop_test.go: TestCountedLoopSupportsOptionalEarlyBreak`](https://github.com/tidbcloud/AgentTeam/src/branch/main/internal/workflow/run/counted_agentteam_test.go) 覆盖。

<!-- slot: 异常 -->
### 常见异常与处置

| 现象 | 原因 | 处置 |
| --- | --- | --- |
| `switch` 路由不稳定 | 把 `first_match` 的 case 顺序当作无关信息 | 按业务优先级排序，并保留唯一默认 case |
| join 永远无法编译 | 某分支没有成功到达配对 join | 关闭每条分支，或用显式失败 `end` 表达分支失败 |
| 并行状态冲突 | 两个可同时激活的分支访问同一状态，且至少一个分支写入 | 各自产出结果，在 `join.merge` 中按键合并；只读与只读可以并行 |
| `join any` 选中失败分支 | 把“第一个终态”误写成语义 | 只把第一个成功到达视为 winner |
| 循环次数差一 | 把 `iterations` 当成回边次数 | 按 body traversal 计数，并测试 1 次和 N 次 |
| 作用域未关闭 | `break` 或最终 `continue` 时仍有活跃并行工作 | 重构图，使内部 join 在离开循环前完成 |

<!-- slot: 审批 -->
### 审批

审批可以位于某个分支或循环体中；每次节点访问产生新的 Activation，并由独立 Effect 与冻结 assignee 标识该审批请求。
当前 `request_approval` 没有创作期超时。
并行审批的响应时限、升级路径和最少通过人数属于业务编排或运营策略；不能把 `join.mode: any` 解释成“多数通过”，也不能把取消 losing branch 解释成撤回已送达的审批或消息。

<!-- slot: 外部系统 -->
### 外部系统

并行分支中的 Agent 可能已经对外部系统产生可见副作用。
`join any` 取消 losing branch 只影响后续 Workflow 图工作，无法自动撤销这些副作用。
设计者应为每个外部写入设置业务幂等键，并把需要的反向操作建模为显式补偿或人工操作手册。

<!-- slot: 验收 -->
### 验收

- [ ] 能说出 `first_match`、`all_match` 和 `fork` 的激活差异。
- [ ] 能解释 `join all` 与 `join any` 的成功、失败和取消边界。
- [ ] 并行分支不依赖 callback 到达顺序，也不竞争写入 Run 状态。
- [ ] 能解释 `iterations: 3` 为何是 3 次 body traversal。
- [ ] 本章 3 个完整 Plan 原样编译后均为 `diagnostics=[]`。

<!-- slot: 媒体槽位 -->
### 媒体槽位

<!-- media: fig-12-01-control-scopes.png -->

建议图示并列展示 `switch: all_match`、`fork`、`join all`、`join any` 与 counted loop 的 lineage，并用不同颜色区分分支键和迭代号。

### 小结

- [ ] 条件选择、固定并行和循环都有显式节点与作用域。
- [ ] join 依据 lineage 关联结果，不依据到达时间猜测。
- [ ] 计数循环执行精确次数；提前退出是成功边界，不是超时。

下一章说明如何让 Agent 在可信工作范围内完成一个 Workflow 激活并回调。