package gateway import "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" // 本文件归属会话编排器的**状态查询与状态帧出口**:谁在读「这个会话忙不忙」, // 以及状态迁移怎么交给传输层。它从 session_orchestrator.go 拆出,使那个文件只留 // 消息状态机(Submit / pending / 续跑 / 结算),并保持在「单文件 ≤500 行」的 // 硬约束内(与 session_events.go / session_tail.go 同一理由)。 // StateSink 把每会话状态转移(busy/pending/idle/interrupt)报告给 // 传输层。网关把它接到 WS hub 广播上。 type StateSink func(sessionID, typ string, data map[string]any) // TurnStateSink 是 StateSink 的**带 turnId 变体**(design/15 §2.5.9):帧壳上的 // `turnId` 是客户端按回合两级分桶的依据,而少数帧(`turn_begin`、`steer_ack`) // 必须让客户端知道**是哪一路**,只有 data 不足以表达。 // // 它刻意是**另一个**回调而不是改 StateSink 的签名:既有接线点(server.go、 // 大量测试)的签名因此零改动,且空 turnID 时明确退化为现签名(帧壳上没有 // turnId,与今天逐字相同)。未接线该变体时,带 turnId 的状态帧退化为普通帧 // ——绝不因此丢一条状态迁移(状态是增益,不是回合能否跑完的前提)。 type TurnStateSink func(sessionID, turnID, typ string, data map[string]any) // SetMode 提升某会话所报告的忙碌状态通道(thinking → // executing → subagent)。它是 loop/subagent 钩子可调用的注入点; // 任何非空闲模式都让会话保持忙碌,而 idle 解除忙碌。为 nil 的会话 // 被惰性创建,因此钩子可以在第一条消息之前设模式。 // // 注意它与「在途路」的关系:真正的忙碌由 st.legs 承载(有路在飞就是忙), // mode 只是给客户端看的通道标签。因此 SetMode 不会被并行的第二路顶掉, // 也不会顶掉第二路的忙态。 // // **已知偏离(诚实登记)**:design/15 §2.5.4 要求多路下 mode「取最重的一路」 // (subagent > executing > thinking)。那需要一个**按路**的模式源,而今天 // 提升模式下井的唯一入口是本函数(会话级、且生产路径上暂无调用方)—— 因此本 // 切片如实保留「会话级 mode:新路起于 thinking、全部收尾才回 idle」的既有语义, // 不伪造一个按路的轻/重判定。等模式提升改成按路注入时,这里再实现加权。 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 } activeTurns := len(st.legs) o.mu.Unlock() if mode == ModeIdle { o.emitState(sessionID, EventSessionIdle, busyPayload(sessionID, mode, activeTurns)) return } o.emitState(sessionID, EventSessionBusy, busyPayload(sessionID, mode, activeTurns)) } // 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 } // ActiveTurns 报告该会话当前的在途路数(design/15 §2.6 的 `activeTurns`)。 // 单路时它非 0 即 1,与「忙不忙」同值;并行时它才是真正的路数 —— 恰好是所有路 // 都收尾后才报 0 的那个判据(design/15 §2.5.3)。 func (o *Orchestrator) ActiveTurns(sessionID string) int { o.mu.Lock() defer o.mu.Unlock() st := o.states[o.normalize(sessionID)] if st == nil { return 0 } return len(st.legs) } // 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) } } // emitTurnState 发出一个**帧壳带 turnId** 的状态转移(design/15 §2.5.9)。 // 未接线带 turnId 的变体时退化为普通状态帧(空 turnID 与现签名等价), // 因此未接线的构建与既有测试绝不因此丢掉一条状态迁移。 func (o *Orchestrator) emitTurnState(sessionID, turnID, typ string, data map[string]any) { if o.emitTurn != nil { o.emitTurn(sessionID, turnID, typ, data) return } o.emitState(sessionID, typ, data) } // notifyPersisted 在接线了传输钩子时把刚持久化的消息交给它, // 使网关能广播 `message_persisted` 以进行跨端同步。 func (o *Orchestrator) notifyPersisted(msg history.Message) { if o.onPersisted != nil { o.onPersisted(msg) } }