AskaEth
89e13cf53d
- turnId 由编排器生成、loop Run 复用（硬前提：否则 turn_begin 与流帧各铸一个，客户端两级分桶立即失配） - StateSink 增带 turnId 的变体（未接线时退化为现签名，不丢状态帧） - 上行 chat.parallel（WS chat 帧 + REST 三入口同形；省略 = 逐字不变） - 单会话额度默认 2 / 硬上限 3 / <2 关闭；额度满或未声明 ⇒ 落回既有排队语义 - I1′：checkpoint 归属、历史配对、忙态计数三处从会话级单例升为路级 - session_idle 收紧为「所有在途路收尾且 pending 空」（仅并行时生效；单路逐字不变） - 预留接口（形状定死、无实现无接线）：tool/legfile.go 的 FileClaimer / ErrFileBusy / WithLegID - 顺手修一个真 bug：并行路 ctx 派生自提交请求作用域，REST 同步入口一写响应就把它连坐杀死（改 context.WithoutCancel 派生） - 23 个新用例（含 -race -count=3）；变异 6 条（turnId 复用 / idle 时点 / 并行升级 / checkpoint 键 / 编排器交 id / pending_ack reason） - 契约变更点 10 条待监工升 v0.65（本人未改 Docs/**）
71 lines
3.3 KiB
Go
71 lines
3.3 KiB
Go
package gateway
|
||
|
||
import (
|
||
"context"
|
||
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||
)
|
||
|
||
// 本文件把「同一会话的**串行续跑链**上一次只有一个回合」从一句约定变成可等待的
|
||
// 进程内围栏（design/15 §3 与裁决 13.4.5 的组合要求；design/15 §2.5 落地真并行后，
|
||
// 需要这条护栏的仍只是串行链 —— 显式声明的并行路用**自己的 checkpoint 归属键**
|
||
// 消除同一竞态，因此不接入这条链）。
|
||
//
|
||
// 中断必须**立即**把会话交回空闲并接受新消息（design/15 §3），但被取消的
|
||
// 回合仍在收尾：loop 直到它自己的 Run 返回之前，才把被放弃回合的快照落定
|
||
// 为不可恢复态（`loop/run.go` 的 `abandonCheckpoint`；2026-10-04 起**所有非崩溃退出**
|
||
// 都走它，不再只限取消）。若新回合此刻
|
||
// 就开跑，它可能在该落定之前读到那份**仍可恢复**的旧快照并续接它 —— 调用方
|
||
// 刚发出的那条消息会被整段顶掉，上下文既不最新也不完整（裁决 13.4.5 的
|
||
// 最新性 / 完整性）。
|
||
//
|
||
// 围栏关掉这条竞态窗口，且不阻塞调用方：新消息照样被立即接受、Submit 立即
|
||
// 返回（非阻塞是硬约束），只是它的回合要等上一回合真正返回之后才开始。
|
||
|
||
// turnFence 是**一条路**的启动围栏。wait 是上一串行回合的收尾信号（nil 表示没有
|
||
// 在途的上一回合，无需等待 —— 并行路与续跑路都用 nil）；done 是本路的收尾信号，
|
||
// 在本路返回时关闭，从而放行等它的下一路。
|
||
type turnFence struct {
|
||
wait <-chan struct{}
|
||
done chan struct{}
|
||
}
|
||
|
||
// newTurnFence 以会话当前的收尾信号为先驱建一道新围栏；prev 为 nil 时
|
||
// wait 也为 nil（本回合可立即开始）。
|
||
func newTurnFence(prev chan struct{}) turnFence {
|
||
var wait <-chan struct{}
|
||
if prev != nil {
|
||
wait = prev
|
||
}
|
||
return turnFence{wait: wait, done: make(chan struct{})}
|
||
}
|
||
|
||
// release 关闭本回合的收尾信号，放行正在等它的下一回合。每个回合恰好调用
|
||
// 一次（它在本回合的唯一出口上被 defer）。
|
||
func (f turnFence) release() { close(f.done) }
|
||
|
||
// awaitTurnFence 等到上一回合真正返回，使被取消回合的 checkpoint 收尾先
|
||
// 落定。等待同时盯着本回合自己的 ctx，因此一个迟迟不收敛的上一回合不会把
|
||
// 本回合永远挂住：用户再按一次「停下」即可让它立即结算。
|
||
//
|
||
// 它只负责**等待**，绝不因此跳过本回合的驱动：本回合若在等到之前就已被
|
||
// 取消（连续两次中断 / 网关关闭），驱动本身会以 ctx 错误立即返回。取消因此
|
||
// 始终被 runner 观察到 —— 被取消的回合绝不会在无人知晓的情况下消失。
|
||
func awaitTurnFence(ctx context.Context, f turnFence) {
|
||
if f.wait == nil {
|
||
return
|
||
}
|
||
select {
|
||
case <-f.wait:
|
||
case <-ctx.Done():
|
||
}
|
||
}
|
||
|
||
// runTurnFenced 在上一回合收尾之后驱动本回合（编排器其余部分仍只认识
|
||
// TurnRunner，围栏只在启动处生效）。
|
||
func (o *Orchestrator) runTurnFenced(ctx context.Context, sessionID string, messages []model.Message, env loop.SessionEnv, fence turnFence) (*loop.Result, error) {
|
||
awaitTurnFence(ctx, fence)
|
||
return o.run(ctx, sessionID, messages, env)
|
||
}
|