458 lines
24 KiB
Go
458 lines
24 KiB
Go
package gateway
|
||
|
||
import (
|
||
"context"
|
||
"io"
|
||
"log/slog"
|
||
"sync"
|
||
"time"
|
||
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||
)
|
||
|
||
// 本文件实现会话消息编排器（DevDocs/design/15）。
|
||
//
|
||
// 它刻意不是任务引擎。design/04 §2（task/subagent）决定
|
||
// 一个重活是否被推到后台；本编排器决定
|
||
// 会话忙碌时新的用户消息如何排队与续跑。两层
|
||
// 正交且可组合：一个会话跑一个任务（04），新消息在
|
||
// 运行中途到达（15），编排器把它握在有界待处理队列中，
|
||
// 并在当前回合结束后带着上一个回合的上下文续跑它。
|
||
//
|
||
// 非阻塞是硬约束：Submit/Interrupt 绝不等待运行中的
|
||
// 回合。队列按会话有界，且跨会话隔离，这里不做
|
||
// 任何身份判定 —— 编排器只排序意图/消息，
|
||
// 身份始终留在 channel 的 actor 元数据中（项目红线）。
|
||
|
||
// Mode 是编排器的每会话忙碌状态通道。它被暴露出来，
|
||
// 使 loop/subagent 钩子能把它升到 executing/subagent（一个
|
||
// 组合根之后可能接线的注入点），但编排器自身只用
|
||
// mode != ModeIdle 跟踪抽象的「忙碌」。
|
||
type Mode string
|
||
|
||
const (
|
||
// ModeIdle 意味着该会话没有在飞回合、也没有待处理续跑。
|
||
ModeIdle Mode = "idle"
|
||
// ModeThinking 意味着有回合在飞（起始的初始状态）。
|
||
ModeThinking Mode = "thinking"
|
||
// ModeExecuting 意味着回合正在执行工具/子步骤。
|
||
ModeExecuting Mode = "executing"
|
||
// ModeSubagent 意味着回合被委派给子代理（design/04 §2）。
|
||
ModeSubagent Mode = "subagent"
|
||
)
|
||
|
||
// 经 WS 通道推给前端的状态事件类型见 session_events.go（它们从本文件拆出，
|
||
// 使编排器本体保持在「单文件 ≤500 行」的硬约束内）。
|
||
|
||
const (
|
||
// defaultMaxPending 限定每会话的待处理队列（design/15 §4）。
|
||
defaultMaxPending = 5
|
||
// defaultMaxTail 限定为续跑而保留的对话尾部，使
|
||
// 续跑的上下文是增量的，而不是全量重放（design/15 §2.2、§4）。
|
||
defaultMaxTail = 20
|
||
// 调用方传入空会话 id 时使用 defaultSessionID。
|
||
defaultSessionID = "web"
|
||
)
|
||
|
||
// SubmitResult 是把一个用户回合交给编排器的结果。它
|
||
// 总是不阻塞地返回：Accepted 为 false 仅在回合被
|
||
// 拒绝时（待处理队列满，或空回合），绝不因为回合忙碌。
|
||
type SubmitResult struct {
|
||
// Accepted 报告回合是否被接收。为 false 且带
|
||
// PendingFull 或 Empty 时携带拒绝原因；它绝不意味着「稍后重试」。
|
||
Accepted bool
|
||
// Immediate 报告因会话空闲而立即开始了一个回合。
|
||
Immediate bool
|
||
// Pending 报告该回合排在忙碌回合之后入队了。
|
||
Pending bool
|
||
// QueueDepth 是本次插入后的待处理深度（Immediate 时为 0）。
|
||
QueueDepth int
|
||
// PendingFull 报告回合被拒绝，因为队列已满
|
||
// （design/15 §4：有界待处理，超容量时拒绝并给出提示）。
|
||
PendingFull bool
|
||
// Empty 报告回合因既无文本也无图片而被拒绝。
|
||
Empty bool
|
||
// Duplicate 报告该回合被判为**重复到达**（G4 幂等）：同一个 clientMsgId
|
||
// 此前已经在本会话落过库，因此本次既不落库、也不启动回合。它是「接住了、
|
||
// 但什么都没做」的诚实答案，绝不是拒绝 —— 客户端据此把本地乐观气泡对齐到
|
||
// MessageID，而不必重试、也不会看到第二个回合。
|
||
Duplicate bool
|
||
// MessageID 是本次提交对应的那条用户消息 id：正常路径是刚落库的消息
|
||
// （带幂等键时是编排器预分配的那一个），重复路径是既有那条。省略
|
||
// clientMsgId 的老客户端为 ""。
|
||
MessageID string
|
||
// Parallel 报告本次提交被**升为一条独立的并行路**（design/15 §2.5：显式声明
|
||
// parallel 且并发额度未满）。它总与 Immediate 同时为真 —— 第二路是「立即起」，
|
||
// 不排到队首；额度已满或未声明时它恒为假，本次落回既有排队语义。
|
||
Parallel bool
|
||
// TurnID 是本次**新起的那一路**的权威回合标识（design/15 §2.5.9）：编排器生成、
|
||
// loop 复用，并与该路全部流帧的 turnId 同值。入队（Pending）或被拒时为空串，
|
||
// 因为那时还没有路。
|
||
TurnID string
|
||
// Err 在同步路径上于运行出错时设置；REST 调用方
|
||
// 用它返回 5xx，而不是编造回复。
|
||
Err error
|
||
}
|
||
|
||
// TurnRunner 执行一个 agent 回合并返回其结果。网关把它
|
||
// 接到 loop 加上流式广播上，因此编排器对回合
|
||
// 实际如何驱动保持不可知（它绝不重新实现 loop 引擎）。
|
||
type TurnRunner func(ctx context.Context, sessionID string, messages []model.Message, env loop.SessionEnv) (*loop.Result, error)
|
||
|
||
// StateSink 与它的带 turnId 变体（TurnStateSink）住在 session_state.go，
|
||
// 与「谁在读状态 / 状态帧怎么出去」放在一起。
|
||
//
|
||
// OrchestratorOptions 配置 Orchestrator。Run 与 Log 必需
|
||
// （Log 可以是丢弃型 logger）；Emit 可选（为 nil 时状态事件被丢弃）。
|
||
// MaxPending/MaxTail 非正时回退到默认值。
|
||
type OrchestratorOptions struct {
|
||
Run TurnRunner
|
||
Emit StateSink
|
||
EmitTurn TurnStateSink
|
||
Log *slog.Logger
|
||
MaxPending int
|
||
MaxTail int
|
||
// MaxParallelTurns 是单会话在途路数的上限（design/15 §2.5.7）：
|
||
// 非正 ⇒ 默认 2；1 ⇒ 关闭真并行（恒 1 路，完全回落到 steer-only 现状）；
|
||
// 超过硬上限 3 ⇒ 钳到 3。它由 resolveMaxParallelTurns 解析。
|
||
MaxParallelTurns int
|
||
// History 是可选的持久化聊天历史存储（design/16 §3.3）。
|
||
// 为 nil 时每次 Append 都被跳过，因此编排器无需存储也能运行
|
||
// （测试 / 未接 history 存储构建的网关）。
|
||
History HistoryStore
|
||
// OnPersisted 非 nil 时，会在用户或助手消息写入后立即
|
||
// 带着完整已存消息被调用，使传输层能广播
|
||
// 一个 `message_persisted` 事件用于跨端实时同步（design/16 §4.2）。
|
||
// 为 nil 时丢弃（测试 / 未接传输钩子构建的网关）。
|
||
OnPersisted func(history.Message)
|
||
}
|
||
|
||
// sessionState 是每会话的消息状态（design/15 §2）。每次变更
|
||
// 都在 Orchestrator 互斥锁下发生；实际的 loop 运行在锁外
|
||
// 完成，因此慢的模型调用绝不阻塞 Submit。
|
||
type sessionState struct {
|
||
mode Mode
|
||
// gen 是会话内**世代号的分配源**：每条新路取一个（单调、不复用）。
|
||
// 旧口径里它同时兼任「当前回合」的身份与陈旧判据；并行落地后这两件事
|
||
// 都归 st.legs —— 「这一路还算数吗」＝「它还在 st.legs 里吗」，
|
||
// 因此启动第二路绝不会让第一路变成陈旧（design/15 §2.5.9）。
|
||
gen uint64
|
||
// legs 是该会话**当前在途**的路（design/15 §2.5 的「路」），键为路的世代号。
|
||
// 单路时它至多一条，且其行为与引入多路之前逐字相同；len(legs) 就是
|
||
// 忙态帧里的 activeTurns，也是「所有路都收尾了吗」的判据。
|
||
legs map[uint64]*turnLeg
|
||
// legSeq 是路序号（1 基）的分配源，即 `turn_begin.data.leg`。
|
||
legSeq int
|
||
pending []pendingMsg
|
||
tail []model.Message // 用于上下文续跑的有界对话尾部
|
||
// fence 是当前在途**串行**路的启动围栏（见 session_turn_fence.go）：下一个
|
||
// 串行回合以它为先驱，等本回合真正返回后才开始 —— 因此中断后立即到达的
|
||
// 消息绝不会在被取消回合落定其 checkpoint 之前读到那份旧快照。并行路不接入
|
||
// 这条链（它的归属键与谁都不同，无需等待）。
|
||
fence turnFence
|
||
// clientMsgs 记录本会话已认领的 clientMsgId → 该消息的 id（G4 幂等）。
|
||
// 它只在状态锁下读写，且只是**快路径 + 同进程并发仲裁**：跨重启的判重由
|
||
// chat_messages 上 (session_id, client_msg_id) 的部分唯一索引裁决
|
||
// （单二进制单进程，不存在跨进程并发）。见 session_persist.go。
|
||
clientMsgs map[string]string
|
||
// clientMsgOrder 是 clientMsgs 的插入顺序，用于有界淘汰
|
||
// （maxClientMsgsPerSession）：淘汰只丢快路径，存储仍是权威。
|
||
clientMsgOrder []string
|
||
// pendingSeq 是待处理队列的单调递增序号源（design/15 §2.4 的转向认领键）。
|
||
// 队列是一条只在队尾追加、可能被整体弹出/清空的先进先出队列，因此「第几条」
|
||
// 不稳定；按 seq 认领已注入的消息，才不会被同一时刻的追加或清空误伤。
|
||
pendingSeq uint64
|
||
}
|
||
|
||
// pendingMsg 是一条已排队用户回合。它携带自己的 env，使续跑
|
||
// 在该消息到达时所用的同一记忆隔离边界 —— 且自 design/22 起
|
||
// 同一角色（env.PersonaID）—— 之下运行，并携带产出该回合的
|
||
// 客户端 deviceID（design/16 §4.5），使续跑的助手回复被标上
|
||
// 发送该消息的同一个客户端；以及该回合的行为模式（design/16
|
||
// §5.1），使续跑回复与发起它的用户消息以同一模式
|
||
// 被持久化。
|
||
type pendingMsg struct {
|
||
messages []model.Message
|
||
env loop.SessionEnv
|
||
deviceID string
|
||
mode string
|
||
at time.Time
|
||
// seq 是这条待处理回合在**本会话**队列里的单调递增序号（design/15 §2.4）。
|
||
// 它只在状态锁下分配，是转向注入「注入＝已消费」认领的依据：队列会被追加、
|
||
// 被弹出、被中断清空，序号却能唯一定位这一条。
|
||
seq uint64
|
||
// persisted 是这条待处理回合已落库的用户消息，供它的续跑回合结算时回写
|
||
// 文档抽取结论（design/24 §6.1）：它在入队那一刻就落了库，而那时抽取还没
|
||
// 发生。见 session_document_status.go。
|
||
persisted []history.Message
|
||
// voice 是这条待处理回合的语音收尾回调（协同 C1）：它源自 voice_input 时
|
||
// 非 nil，使续跑回合结算后仍能把音频定向回发起它的那根连接。
|
||
voice VoiceReplyFunc
|
||
}
|
||
|
||
// Orchestrator 拥有每会话的消息状态机。它对并发安全
|
||
// 且非阻塞：Submit/Interrupt 只在变更状态时加锁，然后把
|
||
// 实际运行交给一个 goroutine。它是 chat 路径上「该会话是否
|
||
// 忙碌」的唯一归属者；loop/task/subagent 引擎被复用，而不是重建。
|
||
type Orchestrator struct {
|
||
mu sync.Mutex
|
||
states map[string]*sessionState
|
||
run TurnRunner
|
||
emit StateSink
|
||
emitTurn TurnStateSink
|
||
log *slog.Logger
|
||
history HistoryStore
|
||
onPersisted func(history.Message)
|
||
maxPending int
|
||
maxTail int
|
||
// maxParallelTurns 是单会话在途路数的生效上限（已解析，见
|
||
// resolveMaxParallelTurns）：1 = 关闭真并行（恒 1 路）。
|
||
maxParallelTurns int
|
||
}
|
||
|
||
// NewOrchestrator 从 opts 构建编排器。Run 必需：它是编排器
|
||
// 复用的回合驱动器（loop，绝不是重新实现的引擎）。
|
||
// Run 为 nil 会让每次 Submit 拒绝而非运行，因此错误配置
|
||
// 不可能静默地让会话处于忙碌。测试在此注入假 runner。
|
||
func NewOrchestrator(opts OrchestratorOptions) *Orchestrator {
|
||
maxPending := opts.MaxPending
|
||
if maxPending <= 0 {
|
||
maxPending = defaultMaxPending
|
||
}
|
||
maxTail := opts.MaxTail
|
||
if maxTail <= 0 {
|
||
maxTail = defaultMaxTail
|
||
}
|
||
if opts.Log == nil {
|
||
opts.Log = slog.New(slog.NewTextHandler(io.Discard, nil))
|
||
}
|
||
return &Orchestrator{
|
||
states: make(map[string]*sessionState),
|
||
run: opts.Run,
|
||
emit: opts.Emit,
|
||
emitTurn: opts.EmitTurn,
|
||
log: opts.Log,
|
||
history: opts.History,
|
||
onPersisted: opts.OnPersisted,
|
||
maxPending: maxPending,
|
||
maxTail: maxTail,
|
||
maxParallelTurns: resolveMaxParallelTurns(opts.MaxParallelTurns),
|
||
}
|
||
}
|
||
|
||
// Submit 把一个用户回合交给会话。它绝不阻塞：空闲会话在
|
||
// 后台 goroutine 上启动回合并返回 Immediate；忙碌会话把
|
||
// 回合入队到有界待处理队列并返回 Pending；队列
|
||
// 满时以 PendingFull 拒绝。此处不自动检测打断词 ——
|
||
// 调用方（前端/意图分类器）在用户要求停止时调用 Interrupt，
|
||
// 依 design/15 §3（打断是一种意图，不是身份判定）。
|
||
func (o *Orchestrator) Submit(ctx context.Context, sessionID string, messages []model.Message, env loop.SessionEnv) SubmitResult {
|
||
sr, _ := o.submitInternal(ctx, o.normalize(sessionID), messages, env, false)
|
||
return sr
|
||
}
|
||
|
||
// SubmitSync 是 REST 非流式路径所用的同步变体。空闲会话
|
||
// 把回合跑完并返回 Result；忙碌会话不阻塞地返回
|
||
// Pending 结果，使调用方能以非阻塞 ack 回复
|
||
// （design/15 §2.1）。它绝不等待在飞回合。
|
||
func (o *Orchestrator) SubmitSync(ctx context.Context, sessionID string, messages []model.Message, env loop.SessionEnv) (SubmitResult, *loop.Result) {
|
||
return o.submitInternal(ctx, o.normalize(sessionID), messages, env, true)
|
||
}
|
||
|
||
// submitInternal 是 Submit/SubmitSync 共用的主体。sync 选择
|
||
// 同步回合路径（REST）还是后台 goroutine 路径（WS）。
|
||
func (o *Orchestrator) submitInternal(ctx context.Context, sessionID string, messages []model.Message, env loop.SessionEnv, sync bool) (SubmitResult, *loop.Result) {
|
||
if o.run == nil {
|
||
o.log.Error("orchestrator submit", "err", "no runner configured", "session", sessionID)
|
||
return SubmitResult{Accepted: false}, nil
|
||
}
|
||
if o.messagesEmpty(messages) {
|
||
return SubmitResult{Empty: true}, nil
|
||
}
|
||
|
||
o.mu.Lock()
|
||
st := o.states[sessionID]
|
||
if st == nil {
|
||
st = &sessionState{mode: ModeIdle}
|
||
o.states[sessionID] = st
|
||
}
|
||
deviceID := deviceIDFrom(ctx)
|
||
msgMode := modeFrom(ctx)
|
||
// 两道门（队列满 / clientMsgId 重复）由 admitLocked 在锁内判定；它不广播，
|
||
// 状态事件在这里解锁之后再发。
|
||
a := o.admitLocked(ctx, st, sessionID)
|
||
if a.full {
|
||
rejected := o.rejectedIdentityLocked(ctx, st, sessionID)
|
||
o.mu.Unlock()
|
||
o.emitPendingFull(ctx, sessionID, a.mode, rejected)
|
||
return SubmitResult{PendingFull: true}, nil
|
||
}
|
||
if a.dup {
|
||
o.mu.Unlock()
|
||
o.emitState(sessionID, EventChatDuplicate, map[string]any{"sessionId": sessionID, "clientMsgId": a.hint.ClientMsgID, "messageId": a.hint.MessageID})
|
||
return SubmitResult{Accepted: true, Duplicate: true, MessageID: a.hint.MessageID}, nil
|
||
}
|
||
// 认领成功：把预分配的消息 id 盖进 ctx。空 clientMsgId 时它是零值，ctx 与
|
||
// 加入该字段之前逐字相同。
|
||
ctx = withClientMsgHint(ctx, a.hint)
|
||
if st.mode != ModeIdle {
|
||
// 忙时：**显式**声明并行且并发额度未满 ⇒ 升为第二路（design/15 §2.5.1
|
||
// 的 I1′）。额度判定与登记在同一个临界区（startParallelLocked 在锁内完成
|
||
// 登记），因此并发到达的两条并行消息不可能双双升级、突破上限。
|
||
// 未声明 / 额度已满 / 队列已满一律落回既有排队语义：不报错、不打断。
|
||
if parallelRequested(ctx) && o.canStartParallelLocked(st) {
|
||
return o.startParallelLocked(ctx, st, sessionID, messages, env, deviceID, msgMode, a.hint), nil
|
||
}
|
||
return o.enqueueLocked(ctx, st, sessionID, messages, env, deviceID, msgMode, a.hint), nil
|
||
}
|
||
|
||
// 本回合的围栏以当前在途串行回合的收尾信号为先驱，使中断后来到的这条
|
||
// 消息不会抢在被取消回合落定旧快照之前开跑（session_turn_fence.go）。
|
||
turn := o.seedTurnLocked(ctx, st, sessionID, messages)
|
||
leg, runCtx, cancel := o.beginLegLocked(ctx, st, newTurnID(), false, newTurnFence(st.fence.done))
|
||
st.fence = leg.fence
|
||
// 本次启动后的在途路数：`turn_begin.activeTurns` 与忙态帧**同源**（正在收尾的路
|
||
// 也计入），因此它只能在锁内读一次 —— 解锁之后再读数会随另一路的收尾漂移。
|
||
activeTurns := len(st.legs)
|
||
o.mu.Unlock()
|
||
|
||
// 本回合的转向源（design/15 §2.4）：loop 在**回合内边界**回调它重读转录，把忙时
|
||
// 新到、已入 pending 的用户消息并入**同一回合**的下一次模型调用（I1：绝不新开
|
||
// 第二回合）。它挂在 runCtx 上，因此只作用于本回合；回合结算或被打断后随 ctx 消亡。
|
||
// 它同时带上本路的 turnId，使 `steer_ack` 的帧壳能指明「被哪一路吃进」。
|
||
runCtx = loop.WithSteerSource(runCtx, o.beginSteer(sessionID, leg.gen, leg.turnID))
|
||
|
||
o.emitState(sessionID, EventSessionBusy, busyPayload(sessionID, ModeThinking, activeTurns))
|
||
// 落库 → 按路配对 → 下发 `turn_begin`：与并行路的启动共用同一段，两条启动
|
||
// 路径的帧序因此同形（见 publishLegStart）。
|
||
o.publishLegStart(ctx, sessionID, leg, messages, deviceID, msgMode, env.PersonaID, activeTurns)
|
||
|
||
if sync {
|
||
return o.runImmediateSync(runCtx, sessionID, leg, turn, env, deviceID, msgMode, a.hint.MessageID)
|
||
}
|
||
|
||
go o.goTurn(sessionID, leg, turn, env, runCtx, cancel, deviceID, msgMode)
|
||
return SubmitResult{Accepted: true, Immediate: true, MessageID: a.hint.MessageID, TurnID: leg.turnID}, nil
|
||
}
|
||
|
||
// runImmediateSync 在**调用方** goroutine 上跑完刚启动的那个回合（REST 同步
|
||
// 路径），结算它并把回合结果一并交回调用方。它与 goTurn 是同一轮的两个驱动
|
||
// 方式（前台等待 / 后台续跑），共享 runTurnFenced + completeTurn 这一对，因此
|
||
// 谁驱动都产生同样的持久化与状态迁移。
|
||
func (o *Orchestrator) runImmediateSync(ctx context.Context, sessionID string, leg *turnLeg, turn []model.Message, env loop.SessionEnv, deviceID, mode, msgID string) (SubmitResult, *loop.Result) {
|
||
res, err := o.runTurnFenced(ctx, sessionID, turn, env, leg.fence)
|
||
leg.fence.release()
|
||
if err != nil {
|
||
o.log.Error("orchestrator turn", "err", err, "session", sessionID)
|
||
}
|
||
_ = o.completeTurn(sessionID, leg, res, err, deviceID, mode, env.PersonaID)
|
||
return SubmitResult{Accepted: true, Immediate: true, MessageID: msgID, TurnID: leg.turnID, Err: err}, res
|
||
}
|
||
|
||
// goTurn 在后台 goroutine 上运行单条**路**，并在它结束时
|
||
// 把结果交给 completeTurn，使编排器能续跑队列。
|
||
// deviceID 用发送该回合的客户端标注持久化助手回复
|
||
// （design/16 §4.5），mode 用该回合的行为模式（design/16 §5.1），而 env 是
|
||
// 该回合解析出的角色/隔离环境（design/22 §4.5），因此回复被
|
||
// 归到产出它的角色名下。leg.fence 是本路的启动围栏：它先等上一**串行**回合
|
||
// 收尾，并在本路返回时放行下一路（session_turn_fence.go）。
|
||
func (o *Orchestrator) goTurn(sessionID string, leg *turnLeg, messages []model.Message, env loop.SessionEnv, ctx context.Context, cancel context.CancelFunc, deviceID, mode string) {
|
||
defer cancel()
|
||
defer leg.fence.release()
|
||
res, err := o.runTurnFenced(ctx, sessionID, messages, env, leg.fence)
|
||
if err != nil && ctx.Err() == nil {
|
||
o.log.Error("orchestrator turn", "err", err, "session", sessionID)
|
||
}
|
||
_ = o.completeTurn(sessionID, leg, res, err, deviceID, mode, env.PersonaID)
|
||
}
|
||
|
||
// completeTurn 在一条路结算后运行（它完成、出错或被
|
||
// 取消）。它把助手回复折进会话的续跑尾部，
|
||
// 然后弹出下一条待处理回合并带着该尾部续跑它 —— 因此下一条
|
||
// 消息「记得」先前上下文，而不是从空白开始
|
||
// （design/15 §2.2：增量，而不是全量重放）。**已经失效的路**（被
|
||
// 打断、被取代：不在 st.legs 里了）是空操作。
|
||
// deviceID 是产出回合的客户端 id，mode 是其行为模式，
|
||
// personaID 是它说话时的角色，都用于标注持久化的回复
|
||
// （design/16 §4.5、§5.1；design/22 §7-5）。
|
||
//
|
||
// 它同时是文档抽取结论回写的落点（design/24 §6.1）：本路用户消息的
|
||
// document 列里此前只有引用，装配期解析出的 status/chars/truncated 在这里补上
|
||
// —— 见 session_document_status.go；语音回合的音频回复同样在此收尾（协同 C1，
|
||
// 见 session_voice.go）：只有这里知道这个回合真正产出了什么。
|
||
//
|
||
// 多路口径（design/15 §2.5.3）：本路先把自己从在途路集合摘掉，**再**决定续跑
|
||
// 还是报空闲。续跑仍由结算中的这一路负责（弹队首、起一条新的**串行**路，占一个
|
||
// 额度）；只有在途路全空且队列也空时才报空闲 —— 单路时该条件恒等价于现状。
|
||
func (o *Orchestrator) completeTurn(sessionID string, leg *turnLeg, res *loop.Result, runErr error, deviceID, mode, personaID string) error {
|
||
gen := leg.gen
|
||
res = o.persistTurnReply(sessionID, gen, res, runErr, deviceID, mode, personaID, leg.turnID)
|
||
o.persistDocumentStatus(sessionID, gen, res)
|
||
o.dispatchVoiceReply(sessionID, gen, res)
|
||
o.mu.Lock()
|
||
st := o.states[sessionID]
|
||
if o.legLocked(st, gen) == nil {
|
||
o.mu.Unlock()
|
||
return nil
|
||
}
|
||
delete(st.legs, gen)
|
||
if res != nil && res.Text != "" {
|
||
st.tail = o.boundedTail(concatMessages(st.tail, assistantTailMessage(res.Text, personaID)))
|
||
}
|
||
if len(st.pending) > 0 {
|
||
next := st.pending[0]
|
||
st.pending = st.pending[1:]
|
||
st.tail = o.boundedTail(concatMessages(st.tail, next.messages...))
|
||
// 续跑回合同样带自己的转向源（design/15 §2.4）：严格 FIFO 弹出下一条之后，
|
||
// 期间**仍在 pending** 的后到消息会在本回合的下一个边界被注入，而不是各起一回合。
|
||
// 本路的 Run 已经返回（我们正处在它的结算里），因此被续跑的回合无需再等
|
||
// 任何在途路：围栏只负责放行它后面的那一个。
|
||
runLeg, runCtx, cancel := o.beginLegLocked(context.Background(), st, newTurnID(), false, newTurnFence(nil))
|
||
st.fence = runLeg.fence
|
||
runCtx = loop.WithSteerSource(runCtx, o.beginSteer(sessionID, runLeg.gen, runLeg.turnID))
|
||
// 续跑路同样可能有**别的在途路**（并行的另一路还在飞）：它也照答自己的那条消息，
|
||
// 同样需要「另有一路正在进行」的结构化摘要（design/15 §2.5.6，主人 2026-10-05
|
||
// 裁决：同一类串话、同一个修法）。快照与上面同一个临界区取（st.legs / st.pending），
|
||
// 没有其它在途路时不挂 ⇒ 单路续跑零成本、行为逐字不变。
|
||
if notice, ok := parallelNoticeLocked(st, runLeg.gen); ok {
|
||
runCtx = loop.WithParallelNotice(runCtx, notice)
|
||
}
|
||
// 续跑的这一轮同样有「回合前已落库的用户消息」，因此把它们的文档状态
|
||
// 记到**这一路**上（design/24 §6.1）；语音回调同理（协同 C1：忙时排队
|
||
// 不许把音频丢掉）。
|
||
runLeg.docs = turnDocuments{gen: runLeg.gen, msgs: next.persisted}
|
||
runLeg.voice = voiceTurn{gen: runLeg.gen, sinks: voiceSinksOf(next.voice)}
|
||
depth := len(st.pending)
|
||
activeTurns := len(st.legs)
|
||
resume := append([]model.Message(nil), st.tail...)
|
||
env := next.env
|
||
o.mu.Unlock()
|
||
|
||
o.emitState(sessionID, EventPendingDrain, map[string]any{"sessionId": sessionID, "mode": ModeThinking, "queueDepth": depth, "activeTurns": activeTurns})
|
||
// 续跑也是一条**新路**，因此同样发一次 turn_begin（kind:"turn"）：它的
|
||
// 归属消息就是刚从队首弹出的那条（它在入队时就已落库，id 取自
|
||
// next.persisted）——没有这一帧，客户端就不知道为什么气泡突然开始回答了。
|
||
o.emitTurnBegin(sessionID, runLeg, userMessageIDs(next.persisted), activeTurns)
|
||
// 续跑路刚起，就把刚从队首弹出的那几条消息认领到它名下（design/15 §2.5.9
|
||
// 的两段式第二段）：它们入队那一刻就落了库，而那时还没有路，此刻才谈得上
|
||
// 归属 —— 补写之后它们与该路的助手回复同值。必须在锁外（它写库）。
|
||
o.assignTurnID(sessionID, runLeg.turnID, next.persisted)
|
||
go o.goTurn(sessionID, runLeg, resume, env, runCtx, cancel, next.deviceID, next.mode)
|
||
return nil
|
||
}
|
||
|
||
// 还有别的路在飞 ⇒ **不报空闲**（design/15 §2.5.3 的时点收紧，仅在并行时
|
||
// 与现状不同：单路时此刻 len(st.legs) 必为 0，因此这一分支不可达）。
|
||
if len(st.legs) > 0 {
|
||
o.mu.Unlock()
|
||
return nil
|
||
}
|
||
st.mode = ModeIdle
|
||
o.mu.Unlock()
|
||
|
||
o.emitState(sessionID, EventSessionIdle, busyPayload(sessionID, ModeIdle, 0))
|
||
return nil
|
||
}
|