同步原语 v0.1（draft）

L2 只做四种 multiport cell：Join(k)、Select(k)、Barrier(k)、Gate。目标：足够表达 Agent 协作（fan-in/race/barrier/approval），又不把完整 multiport/π 演算的复杂度带进 v1。

1. Join(k)

形态

Join(k) : (T₁, …, Tₖ) → T

k 个 principal port（输入），1 个 auxiliary port（输出）。
只有当 k 个输入端口都连接了消息时，才形成活跃配置。
点火：消费 k 条消息，产生一个结果（通常是 tuple 或用户函数 f(t₁…tₖ)）。

规则

Join(k) ⋈ { m₁, …, mₖ }  =>  Send(out, f(m₁,…,mₖ))

（f 在创建 Join 时绑定；默认是 tuple 构造。）

类型

Γ ⊢ mᵢ : Tᵢ    f : T₁ × … × Tₖ → T
------------------------------------
Γ ⊢ Join(k, f) : (T₁, …, Tₖ) → T

输入端口是线性的：每个只消费一次。
输出端口也是线性的。

用途

fan-in / map-reduce 的 reduce 阶段。
等待多个工具/模型结果。
事务性组合：所有输入到齐才继续。

2. Select(k)

形态

Select(k) : (T₁, …, Tₖ) → T ⊕ Cancelled

k 个 principal 输入；第一个到达的消息获胜。
其余输入收到 Cancel（如果上游声明了 cancel 能力）；未声明的上游被擦除（ε），但其 effect 可能仍在运行（记账/丢弃结果）。

规则

Select(k) ⋈ mᵢ  =>  Send(out, mᵢ) | Cancel(mⱼ) for j ≠ i

类型

Γ ⊢ mᵢ : T     ∀j. Γ ⊢ Cancel(mⱼ) ok
------------------------------------
Γ ⊢ Select(k) : (T, …, T) → T

所有输入类型必须一致（或可上转为共同类型）。
上游若是 Effect/Agent，必须有 cancel : ?Cancel 端口；否则类型错误。

用途

race / 超时 / 多模型投票（第一个成功）。
投机执行：先返回的分支获胜，其他取消。

取消协议

Select 向败者发送 Cancel。
败者（Effect/Agent）收到后：
- 若 provider 支持取消 → abort；发送 Cancelled 回执。
- 否则 → 标记 Cancelled，结果到达后丢弃（仍计费）。
Select 不等待回执（避免阻塞）；回执用于账本/审计。

3. Barrier(k)

形态

Barrier(k) : (Signal, …, Signal) → (Signal, …, Signal)

k 个 principal 输入；收齐后向 k 个 auxiliary 输出各发一个 Signal。
用于阶段同步：所有 worker 完成上一阶段后才进入下一阶段。

规则

Barrier(k) ⋈ { s₁, …, sₖ }  =>  Send(out₁, s) | … | Send(outₖ, s)

用途

批量推理：等所有分片完成再聚合。
训练/评估的 epoch 边界（如果用于 Agent 编排）。

注意

Barrier 是全局同步点；会降低并行度。只在语义需要时使用。
死锁：如果某个输入永远不到，Barrier 永远等待；运行时用 deadline/timeout 兜底。

4. Gate

形态

Gate : ?Token → !Token

一个 principal 输入；只有 open 状态才转发消息。
open/close 由控制消息或 supervisor 设置。
用于人工审批、限流、熔断。

规则

Gate[Open] ⋈ m        =>  Send(out, m)
Gate[Closed] ⋈ m      =>  Gate[Closed] | Queue(m)   // 或拒绝

v1 提供两种策略：queue（有界队列）与 reject（返回 Error<Closed>）。

5. 死锁与类型安全

保守检查（v1）

构造 Join/Select/Barrier 时，检查输入端口的上游是否能到达一个"最终会发送"的 cell。
拒绝明显的环：Join → … → Join 且没有外部输入/超时。
对递归协议，要求有一个 base case 能打破等待。

运行时兜底

每个 Join/Select/Barrier 有 deadline（默认继承任务 deadline；否则 5 分钟）。
超时 → 注入 Error<Timeout> 到输出；未消费的输入被取消/擦除。
记录 E-DEADLOCK 事件（trace），供调试。

定理（待证明）

对 well-typed、树形（无环等待）的 Join/Select/Barrier 网，若所有上游最终产生消息或错误，则网不会卡死；最终结果与同步顺序无关（合流）。

证明思路：把每个同步 cell 看作一个 Petri 网 transition；树形结构保证偏序无环；类型保证每个输入恰好一个生产者。完整证明在 specs/ain-sync-proof.md（待写）。

6. 实现

表示

Join(k):  header | principal[0..k-1] | aux_out | state{ received bitmask, values[] }
Select(k): header | principal[0..k-1] | aux_out | state{ winner, cancelled[] }
Barrier(k): header | principal[0..k-1] | aux_out[0..k-1] | state{ received }
Gate: header | principal | aux_out | state{ open|closed, queue }

每个 principal 端口是一个原子 wire；到达时 CAS 登记到 received。
当 received 满 → 形成活跃配置 → 入 redex 队列（claim 整个 cell）。
与二元 cell 不同，multiport cell 的"活跃对"是一个集合；调度器需要 claim_cell（不是 claim_pair）。

调度

Join/Select/Barrier 的 cell 有 Busy 位；只有 owner 能点火。
到达消息与点火可能并发：用原子 bitmask + CAS；点火前再次确认所有输入仍在。
取消：Select 点火后向败者发消息；败者的 cancel 处理是普通规则。

成本

Join(k) 的点火成本 = k（消费 k 条消息）+ f 的成本。
Select(k) 的期望成本 = 1 次点火 + (k-1) 次取消（若上游支持）。
Barrier(k) = k 次接收 + k 次发送。

7. 示例

fan-out / join

flow MapReduce {
  input items: List<Item>
  output result: Result

  par {
    branch a = Worker.map(items[0..n/2])
    branch b = Worker.map(items[n/2..n])
  } -> join { a, b } as parts

  Reducer.reduce(parts) -> result
}

race

flow Fastest {
  input q: Question
  output a: Answer

  race {
    M1.answer(q)
    M2.answer(q)
  } -> a
}

barrier

flow Batch {
  input xs: List<Item>
  output rs: List<Result>

  par { branch chunks = split(xs, 8) }
  barrier(8)
  par { branch results = map(chunks, Worker.run) }
  results -> rs
}

8. 版本

v0.1：Join/Select/Barrier/Gate；无动态 arity；无通道名传递。
v0.2（计划）：动态 k（运行时创建）、Select 的加权/超时分支、Gate 的令牌桶。
v1.0：稳定语义 + 证明 + 完整测试向量。

在 GitHub 编辑本文