Files
Cyrene_v2/internal/services/gateway/session_orchestrator.go
T
AskaEth 074ffc4ebf
CI / Backend (build / vet / test / cross-compile) (push) Has been cancelled
CI / Frontend (tsc + vite build) (push) Has been cancelled
docs(gateway): gateway 剩余文件注释改中文(逐行 1:1,代码零字节改动)
主人 2026-10-03 拍板「注释一律中文」;gateway 至此整包注释中文化完成(本会话共 3 批:
已推库 16 个文件 + 本批剩余文件)。
校验:机械校验(非注释改动行)= 空;剥注释后代码文本逐行比对全等(不一致文件数 0);
全树 go test ./... 58 包 ok / 0 FAIL;交叉编译绿。
2026-10-03 22:28:25 +08:00

476 lines
18 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
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 通道推给前端的状态事件类型。每个的 data map 都
// 携带 sessionId,使客户端能按会话关联状态。
const (
EventSessionBusy = "session_busy"
EventPendingAck = "pending_ack"
EventPendingDrain = "pending_drain"
EventSessionIdle = "session_idle"
EventSessionInterrupt = "session_interrupted"
EventPendingFull = "pending_full"
)
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
// 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 把每会话状态转移(busy/pending/idle/interrupt)报告给
// 传输层。网关把它接到 WS hub 广播上。
type StateSink func(sessionID, typ string, data map[string]any)
// OrchestratorOptions 配置 Orchestrator。Run 与 Log 必需
// (Log 可以是丢弃型 logger);Emit 可选(为 nil 时状态事件被丢弃)。
// MaxPending/MaxTail 非正时回退到默认值。
type OrchestratorOptions struct {
Run TurnRunner
Emit StateSink
Log *slog.Logger
MaxPending int
MaxTail 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 uint64 // generation:在打断时以及新的顶层回合时递增
cancel context.CancelFunc
pending []pendingMsg
tail []model.Message // 用于上下文续跑的有界对话尾部
}
// 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
}
// Orchestrator 拥有每会话的消息状态机。它对并发安全
// 且非阻塞:Submit/Interrupt 只在变更状态时加锁,然后把
// 实际运行交给一个 goroutine。它是 chat 路径上「该会话是否
// 忙碌」的唯一归属者;loop/task/subagent 引擎被复用,而不是重建。
type Orchestrator struct {
mu sync.Mutex
states map[string]*sessionState
run TurnRunner
emit StateSink
log *slog.Logger
history HistoryStore
onPersisted func(history.Message)
maxPending int
maxTail 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,
log: opts.Log,
history: opts.History,
onPersisted: opts.OnPersisted,
maxPending: maxPending,
maxTail: maxTail,
}
}
// 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)
if st.mode != ModeIdle {
if len(st.pending) >= o.maxPending {
mode := st.mode
o.mu.Unlock()
o.emitState(sessionID, EventPendingFull, map[string]any{"sessionId": sessionID, "mode": mode, "max": o.maxPending})
return SubmitResult{PendingFull: true}, nil
}
st.pending = append(st.pending, pendingMsg{messages: messages, env: env, deviceID: deviceID, mode: msgMode, at: time.Now()})
depth := len(st.pending)
mode := st.mode
o.mu.Unlock()
o.persistUserMessages(ctx, sessionID, messages, deviceID, msgMode, env.PersonaID)
o.emitState(sessionID, EventPendingAck, map[string]any{"sessionId": sessionID, "mode": mode, "queueDepth": depth})
return SubmitResult{Accepted: true, Pending: true, QueueDepth: depth}, nil
}
st.gen++
gen := st.gen
st.mode = ModeThinking
// 新回合延续该会话的对话:持久化转录被播种在
// 刚发送的消息之前(design/15 §2.2/§4,design/22
// §7-5),且其助手话语保持说话时的角色。该读取
// 在此发生,在下面 persistUserMessages 写入新消息之前,因此刚
// 发送的消息绝不在上下文中重复。没有历史(或空会话)的
// 存储让该回合只使用调用方给出的消息。它是在状态锁下
// 的有界索引读取(transcriptContextCount 行),而状态锁
// 从不会被运行中的回合持有,因此它无法把 Submit 堵在回合后面。
turn := o.startTurnMessages(ctx, sessionID, messages)
st.tail = o.boundedTail(turn)
runCtx, cancel := context.WithCancel(ctx)
st.cancel = cancel
o.mu.Unlock()
o.emitState(sessionID, EventSessionBusy, map[string]any{"sessionId": sessionID, "mode": ModeThinking})
// 只有此刻到达的消息会被持久化:已播种的转录
// 已经持久,重新追加它会重复整个对话。
o.persistUserMessages(ctx, sessionID, messages, deviceID, msgMode, env.PersonaID)
if sync {
res, err := o.run(runCtx, sessionID, turn, env)
if err != nil {
o.log.Error("orchestrator turn", "err", err, "session", sessionID)
}
_ = o.completeTurn(sessionID, gen, res, err, deviceID, msgMode, env.PersonaID)
return SubmitResult{Accepted: true, Immediate: true, Err: err}, res
}
go o.goTurn(sessionID, gen, turn, env, runCtx, cancel, deviceID, msgMode)
return SubmitResult{Accepted: true, Immediate: true}, nil
}
// goTurn 在后台 goroutine 上运行单个回合,并在它结束时
// 把结果交给 completeTurn,使编排器能续跑队列。
// deviceID 用发送该回合的客户端标注持久化助手回复
// (design/16 §4.5),mode 用该回合的行为模式(design/16 §5.1),而 env 是
// 该回合解析出的角色/隔离环境(design/22 §4.5),因此回复被
// 归到产出它的角色名下。
func (o *Orchestrator) goTurn(sessionID string, gen uint64, messages []model.Message, env loop.SessionEnv, ctx context.Context, cancel context.CancelFunc, deviceID, mode string) {
defer cancel()
res, err := o.run(ctx, sessionID, messages, env)
if err != nil && ctx.Err() == nil {
o.log.Error("orchestrator turn", "err", err, "session", sessionID)
}
_ = o.completeTurn(sessionID, gen, res, err, deviceID, mode, env.PersonaID)
}
// completeTurn 在一个回合结算后运行(它完成、出错或被
// 取消)。它把助手回复折进会话的续跑尾部,
// 然后弹出下一条待处理回合并带着该尾部续跑它 —— 因此下一条
// 消息「记得」先前上下文,而不是从空白开始
// (design/15 §2.2:增量,而不是全量重放)。陈旧的代次(一次
// 打断或一个更新的顶层回合取代了本次)是空操作。
// deviceID 是产出回合的客户端 id,mode 是其行为模式,
// personaID 是它说话时的角色,都用于标注持久化的回复
// (design/16 §4.5、§5.1;design/22 §7-5)。
func (o *Orchestrator) completeTurn(sessionID string, gen uint64, res *loop.Result, runErr error, deviceID, mode, personaID string) error {
o.persistAssistantReply(sessionID, res, deviceID, mode, personaID)
o.mu.Lock()
st := o.states[sessionID]
if st == nil || st.gen != gen {
o.mu.Unlock()
return nil
}
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...))
st.mode = ModeThinking
st.gen++
gen = st.gen
runCtx, cancel := context.WithCancel(context.Background())
st.cancel = cancel
depth := len(st.pending)
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})
go o.goTurn(sessionID, gen, resume, env, runCtx, cancel, next.deviceID, next.mode)
return nil
}
st.mode = ModeIdle
st.cancel = nil
o.mu.Unlock()
o.emitState(sessionID, EventSessionIdle, map[string]any{"sessionId": sessionID, "mode": ModeIdle})
return nil
}
// Interrupt 取消在飞回合、清空待处理队列并把会话
// 回到空闲(design/15 §3,「停下/算了/换个话题」路径)。它是忙碌
// 会话不必排空队列就能复位的唯一方式;运行中 goroutine 的
// 完成变成陈旧代次,不续跑任何东西。它返回
// 被丢弃的待处理回合数。
func (o *Orchestrator) Interrupt(sessionID string) int {
return o.interrupt(o.normalize(sessionID))
}
// InterruptAll 取消每一个在飞会话并清空所有待处理队列。它
// 在网关关闭时使用,使没有后台回合比 hub 活得更久。它返回
// 被丢弃的待处理回合总数。
func (o *Orchestrator) InterruptAll() int {
o.mu.Lock()
total := 0
for id, st := range o.states {
total += o.interruptLocked(id, st)
}
o.mu.Unlock()
return total
}
func (o *Orchestrator) interrupt(sessionID string) int {
o.mu.Lock()
defer o.mu.Unlock()
return o.interruptLocked(sessionID, o.states[sessionID])
}
func (o *Orchestrator) interruptLocked(sessionID string, st *sessionState) int {
if st == nil || st.mode == ModeIdle {
return 0
}
if st.cancel != nil {
st.cancel()
}
dropped := len(st.pending)
st.pending = nil
st.gen++
st.mode = ModeIdle
st.cancel = nil
o.emitState(sessionID, EventSessionInterrupt, map[string]any{"sessionId": sessionID, "mode": ModeIdle, "dropped": dropped})
return dropped
}
// SetMode 提升某会话所报告的忙碌状态通道(thinking →
// executing → subagent)。它是 loop/subagent 钩子可调用的注入点;
// 任何非空闲模式都让会话保持忙碌,而 idle 解除忙碌。为 nil 的会话
// 被惰性创建,因此钩子可以在第一条消息之前设模式。
func (o *Orchestrator) SetMode(sessionID string, mode Mode) {
sessionID = o.normalize(sessionID)
o.mu.Lock()
st := o.states[sessionID]
if st == nil {
st = &sessionState{mode: mode}
o.states[sessionID] = st
} else {
st.mode = mode
}
o.mu.Unlock()
if mode == ModeIdle {
o.emitState(sessionID, EventSessionIdle, map[string]any{"sessionId": sessionID, "mode": mode})
return
}
o.emitState(sessionID, EventSessionBusy, map[string]any{"sessionId": sessionID, "mode": mode})
}
// Busy 报告该会话当前是否有在飞回合。
func (o *Orchestrator) Busy(sessionID string) bool {
o.mu.Lock()
defer o.mu.Unlock()
st := o.states[o.normalize(sessionID)]
return st != nil && st.mode != ModeIdle
}
// QueueDepth 报告某会话的待处理回合数。
func (o *Orchestrator) QueueDepth(sessionID string) int {
o.mu.Lock()
defer o.mu.Unlock()
st := o.states[o.normalize(sessionID)]
if st == nil {
return 0
}
return len(st.pending)
}
// ModeOf 报告某会话当前的忙碌状态通道(未知时为 idle)。
func (o *Orchestrator) ModeOf(sessionID string) Mode {
o.mu.Lock()
defer o.mu.Unlock()
st := o.states[o.normalize(sessionID)]
if st == nil {
return ModeIdle
}
return st.mode
}
// normalize 把空会话 id 映射为默认值,使状态 map 对省略它的
// 调用方保持稳定(Web 主会话)。
func (o *Orchestrator) normalize(sessionID string) string {
if sessionID == "" {
return defaultSessionID
}
return sessionID
}
// emitState 在接线了传输层时把状态转移转发给它。
func (o *Orchestrator) emitState(sessionID, typ string, data map[string]any) {
if o.emit != nil {
o.emit(sessionID, typ, data)
}
}
// notifyPersisted 在接线了传输钩子时把刚持久化的消息交给它,
// 使网关能广播 `message_persisted` 以进行跨端同步。
func (o *Orchestrator) notifyPersisted(msg history.Message) {
if o.onPersisted != nil {
o.onPersisted(msg)
}
}
// deviceIDContextKey 是携带消息客户端 deviceID 的 context 键
// (design/16 §4.5)从 HTTP/WS 边界引入编排器的持久化。
type deviceIDContextKey struct{}
// withDeviceID 返回携带 deviceID 的 context,使编排器能标注
// 持久化的用户消息。空 deviceID 是空操作(不写入标签)。
func withDeviceID(ctx context.Context, deviceID string) context.Context {
return context.WithValue(ctx, deviceIDContextKey{}, deviceID)
}
// deviceIDFrom 返回 ctx 携带的 deviceID,未设置时返回 ""。它
// 只返回格式良好的非空值;其他一律视为缺失。
func deviceIDFrom(ctx context.Context) string {
v, _ := ctx.Value(deviceIDContextKey{}).(string)
return v
}