AskaEth
3a01b4e236
- G5：pending_full 增 clientMsgId/messageId（仅在可给出时出现、绝不新铸 id —— 队列满的拒绝发生在幂等键认领之前，代铸会让客户端重试被判重复而永远进不来）；另定向回发送者 {type:error,data:{code:chat_busy,...}}（与 REST 409 chat_busy 同码），广播帧照旧带同样的键供其它端对齐。
- C1：语音回合改走与文本同一条路径（编排器 Submit → 落库 → 流式 → 跨端同步），deviceId 走既有 withDeviceID、platform 走存储默认 web；未接编排器时保留旧直连路径。音频在回合结算处（persistAssistantReply → persistDocumentStatus → dispatchVoiceReply）合成并定向回原连接，回调各自 goroutine（TTS 延迟不拖住 pending_drain/session_idle）；续跑 FIFO 与转向注入两条路径都带走回调 ⇒ 忙时排队/被注入的语音也不丢音频。
- C3：voice_output 增 sessionId/deviceId；三处主动加码（按最优方向）：① voice_input 增可选 clientMsgId（语音也有幂等）② 不再拿 stopReason 顶替空正文当语音内容（空正文走 provider 降级信号）③ TTS 失败照旧回帧 + provider:silent（不静默）。
- 拆 chat_documents_test.go（510 行越界）→ 373 + 160，只搬不改（拆前后同 -run 过滤 PASS 集合逐条一致）。
- 主会话独立验收：隔离 worktree（HEAD + 本批次 11 文件）build/vet 绿、gateway -race 绿、全树 59 包（唯一红是既有负载 flake orchestration/TestSpawnTimeoutFromToolDefault）、linux+darwin 交叉编译绿；变异独立复现（pending_full 摘掉 clientMsgId ⇒ 3 条红）。
- 待办（已登记）：Docs/api-client.md 与 api-reference.md 的帧/字段登记随 v0.59 一并落笔（本批次刻意未动 Docs，避免与契约批次打架）。
199 lines
8.9 KiB
Go
199 lines
8.9 KiB
Go
package gateway
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/auth"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/orchestration"
|
||
)
|
||
|
||
// wsForbiddenCode 是非 owner/admin 连接发送任何上行指令时收到的帧的机器可读
|
||
// 错误码（design/11 §7.2 B）。它沿用本文件既有的错误帧形状 ——
|
||
// wsEvent{Type: "error", Data: map[string]any} —— 客户端用它已经认识的 code
|
||
// 解码，仅凭这个键就能把「你没有这个权限」与处理函数级失败区分开。
|
||
const wsForbiddenCode = "forbidden"
|
||
|
||
// route 按类型分发一个已解码的上行（客户端→服务端）帧。
|
||
//
|
||
// 角色门（design/11 §7.2 B，主人于 2026-10-03 决策）：非 owner/admin 连接
|
||
// 不得发送任何应用内指令。该检查是本函数的第一条语句，且默认拒 —— 它刻意
|
||
// 不是一张「禁止类型」清单 —— 因此后续里程碑新增的帧类型在有人显式登记
|
||
// 之前一律被拒。viewer 的 /ws/chat 连接因此是只读监视器：它跟随会话状态机，
|
||
// 但永远无法驱动模型、取消回合、读取子代理上下文或任务输出，也不能输入语音。
|
||
// 每个被拒的帧只在那一根连接上得到 {type:"error", data:{code:"forbidden"}}
|
||
// 的应答；不广播、不进入任何处理函数，因此不产生模型调用，也不落库。
|
||
//
|
||
// 放宽此门：viewer 侧的上行帧只有先在设计里给它开口、并登记到 design/11 §7.2
|
||
// 之后才允许（「必须先在设计里给它开口并登记白名单」）。下方的一律拒绝正是
|
||
// 让「忘记这一步」仍然安全的保障，所以放开某一个类型意味着把它换成一张
|
||
// 登记表，而不是删掉它。
|
||
//
|
||
// 协议级帧（ping/pong/close）不会到达这里 —— gorilla 在 ReadMessage 内部
|
||
// 处理 —— 因此这道门不会杀死连接。完全不是 JSON 的帧由 readPump 在分发前
|
||
// 应答；它不带类型、触发不了任何东西，所以保留解析层的应答。
|
||
func (c *wsConn) route(in wsInbound) {
|
||
if !ownerAdminClaims(c.claims) {
|
||
c.h.log.Debug("ws inbound rejected", "type", in.Type, "role", claimRole(c.claims))
|
||
c.enqueueJSON(wsEvent{Type: "error", Data: map[string]any{"code": wsForbiddenCode}})
|
||
return
|
||
}
|
||
switch in.Type {
|
||
case "mode_switch":
|
||
c.publishPublic("ws.mode_switch", rawData(in.Data))
|
||
case "chat":
|
||
// design/15 §2.1：会话消息「不能被 drop」——bus 的有界订阅缓冲满时
|
||
// 会丢弃事件（那是给事件/通知用的通道），而用户消息被丢就是上下文
|
||
// 不完整。因此上行 chat 直接进入会话编排器的非阻塞 Submit（裁决
|
||
// 13.4.5：忙时入 pending、续接更新的旧上下文），不经 bus。
|
||
c.submitChat(rawData(in.Data))
|
||
case "interrupt":
|
||
// design/15 §3: 用户说「停下/算了」→ 取消在飞回合；待处理队列按
|
||
// data.scope 决定去留（协同项 A2：省略/"all" = 连队列一起清，
|
||
// "current" = 只取消当前回合并保留队列）。非法 scope 由连接级
|
||
// 处理函数定向回 bad_request，因此这里调连接自己的方法。
|
||
c.handleInterrupt(in.Data)
|
||
case "voice_input":
|
||
// 语音经注入的 STT→chat→TTS 管线处理。没有 voice service 时回一个
|
||
// 明确的错误，而不是把帧丢掉。转写之后的**回合**与文本同一条路（协同
|
||
// C1：编排器 → 落库 → 流式 → 跨端同步），音频回复在回合结算时作为
|
||
// voice_output 事件定向回发起请求的那根连接。
|
||
c.handleVoiceInput(in)
|
||
case "subagent_ask":
|
||
c.handleSubagentAsk(in)
|
||
case "task_output_query":
|
||
c.handleTaskOutputQuery(in)
|
||
default:
|
||
c.enqueueJSON(wsEvent{Type: "error", Data: map[string]any{"message": "unknown event type"}})
|
||
}
|
||
}
|
||
|
||
// subagentAskRequest 是 subagent_ask 帧的解码载荷。Mode 选择审计窗口
|
||
// （""/full/tail/keyword）；n 界定 tail 查询的条数；query 是 keyword 查询的
|
||
// 关键词。
|
||
type subagentAskRequest struct {
|
||
ID string `json:"id"`
|
||
Mode string `json:"mode"`
|
||
N int `json:"n"`
|
||
Query string `json:"query"`
|
||
}
|
||
|
||
// taskOutputRequest 是 task_output_query 帧的解码载荷。
|
||
type taskOutputRequest struct {
|
||
TaskID string `json:"taskId"`
|
||
}
|
||
|
||
// handleSubagentAsk 审计某个子代理的上下文，并以携带真实上下文窗口的
|
||
// ws.subagent_event 帧回复。未注入 auditor（gateway 没接 scheduler）时回一个
|
||
// 明确的错误，而不是编造一个 "pending" 快照。
|
||
func (c *wsConn) handleSubagentAsk(in wsInbound) {
|
||
if c.h.auditor == nil {
|
||
c.enqueueJSON(wsEvent{Type: "error", Data: map[string]any{"message": "subagent audit unavailable"}})
|
||
return
|
||
}
|
||
var req subagentAskRequest
|
||
_ = json.Unmarshal(in.Data, &req)
|
||
if req.ID == "" {
|
||
c.enqueueJSON(wsEvent{Type: "error", Data: map[string]any{"message": "subagent id required"}})
|
||
return
|
||
}
|
||
mode := orchestration.AuditMode(req.Mode)
|
||
if mode == "" {
|
||
mode = orchestration.AuditFull
|
||
}
|
||
res, err := c.h.auditor.AuditContext(context.Background(), req.ID, mode, req.N, req.Query)
|
||
if err != nil {
|
||
c.h.log.Debug("subagent audit", "id", req.ID, "err", err)
|
||
c.enqueueJSON(wsEvent{Type: "error", Data: map[string]any{"message": err.Error()}})
|
||
return
|
||
}
|
||
c.enqueueJSON(wsEvent{Type: "subagent_event", Data: res})
|
||
}
|
||
|
||
// handleTaskOutputQuery 取回 taskId 的真实任务输出，并以 ws.task_output 帧回复。
|
||
// 未注入 task store 时回一个明确的错误，而不是空的 {}。仍在运行的任务返回迄今
|
||
// 已捕获的输出。
|
||
func (c *wsConn) handleTaskOutputQuery(in wsInbound) {
|
||
if c.h.taskOutputs == nil {
|
||
c.enqueueJSON(wsEvent{Type: "error", Data: map[string]any{"message": "task output unavailable"}})
|
||
return
|
||
}
|
||
var req taskOutputRequest
|
||
_ = json.Unmarshal(in.Data, &req)
|
||
if req.TaskID == "" {
|
||
c.enqueueJSON(wsEvent{Type: "error", Data: map[string]any{"message": "task id required"}})
|
||
return
|
||
}
|
||
ev, err := c.h.taskOutputs.TaskOutput(context.Background(), req.TaskID)
|
||
if err != nil {
|
||
c.h.log.Debug("task output query", "id", req.TaskID, "err", err)
|
||
c.enqueueJSON(wsEvent{Type: "error", Data: map[string]any{"message": err.Error()}})
|
||
return
|
||
}
|
||
text := ev.Output.Stdout
|
||
if ev.Output.Stderr != "" {
|
||
if text != "" {
|
||
text += "\n"
|
||
}
|
||
text += ev.Output.Stderr
|
||
}
|
||
c.enqueueJSON(wsEvent{Type: "task_output", Data: map[string]any{
|
||
"taskId": ev.ID,
|
||
"output": text,
|
||
"status": string(ev.Status),
|
||
}})
|
||
}
|
||
|
||
// claimRole 为上行业务拒绝的日志记录标注连接角色。claims 为 nil 的连接
|
||
// （单元测试构造的 fail-closed 裸连接）记为 "none"，而不是解引用空指针。
|
||
func claimRole(claims *auth.Claims) string {
|
||
if claims == nil {
|
||
return "none"
|
||
}
|
||
return string(claims.Role)
|
||
}
|
||
|
||
// wsChatBusyCode 是待处理队列满时定向回发送者的机器可读错误码。它与 REST
|
||
// 同步/流式路径的 409 `chat_busy` **同码**：客户端用同一个 code 处理两条入口的
|
||
// 「队列满」，不必为 WS 再造一套判定（协同 G5）。
|
||
const wsChatBusyCode = "chat_busy"
|
||
|
||
// notifyPendingFull 定向回一条拒绝帧，告诉发送这条消息的连接「会话正忙、待处理
|
||
// 消息已满、本条未入队」，并带上能对齐本地乐观气泡的键（协同 G5）：clientMsgId
|
||
// 原样带回（未携带该键时不伪造），messageId 只在该键此前已落库/已认领时出现。
|
||
//
|
||
// 它只到发送者、不广播：广播的 pending_full 是状态信号，其余连接按 clientMsgId
|
||
// 自行忽略即可。
|
||
func (c *wsConn) notifyPendingFull(sessionID, clientMsgID, messageID string) {
|
||
data := map[string]any{"code": wsChatBusyCode, "message": "会话正忙，待处理消息已满，本条未入队", "sessionId": sessionID}
|
||
if clientMsgID != "" {
|
||
data["clientMsgId"] = clientMsgID
|
||
}
|
||
if messageID != "" {
|
||
data["messageId"] = messageID
|
||
}
|
||
c.enqueueJSON(wsEvent{Type: "error", Data: data})
|
||
}
|
||
|
||
// submitChat 把一条 chat 上行帧直接交给会话编排器（design/15 §2.1/§2.2），
|
||
// 而不是经 bus 转发：会话消息不得依赖可丢弃的通道（见 route 的 chat 分支）。
|
||
// 解码、角色解析与用户消息落库都在该 goroutine 上完成，因此读泵绝不等待
|
||
// 回合（design/15 §4「不阻塞是硬约束」）。
|
||
//
|
||
// 未接编排器时（单元测试构造的裸 hub、或未接编排器的构建）回退到既有的
|
||
// bus 触发器路径，使 hub 的 ws.chat 契约逐字节不变。
|
||
func (c *wsConn) submitChat(payload any) {
|
||
if c.h == nil {
|
||
return
|
||
}
|
||
if c.h.orchestrator == nil {
|
||
c.publishPublic(wsChatSubject, payload)
|
||
return
|
||
}
|
||
// 回合自身的取消由编排器持有（Interrupt/InterruptAll 取消它），
|
||
// 因此这里的基 ctx 只需是存活到回合结束的背景上下文。队列满的拒绝要定向回
|
||
// **这条**连接（协同 G5）：编排器只认 ctx 上给出的回调（REST 走 409，不接）。
|
||
ctx := withPendingFullNotifier(context.Background(), c.notifyPendingFull)
|
||
go c.h.streamChat(ctx, payload)
|
||
}
|