AskaEth
074ffc4ebf
主人 2026-10-03 拍板「注释一律中文」；gateway 至此整包注释中文化完成（本会话共 3 批： 已推库 16 个文件 + 本批剩余文件）。 校验：机械校验（非注释改动行）= 空；剥注释后代码文本逐行比对全等（不一致文件数 0）； 全树 go test ./... 58 包 ok / 0 FAIL；交叉编译绿。
126 lines
4.7 KiB
Go
126 lines
4.7 KiB
Go
package gateway
|
||
|
||
import (
|
||
"strings"
|
||
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/bus"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/orchestration"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/orchestration/comm"
|
||
)
|
||
|
||
// eventBridge 把生产者 bus 主题映射为其下行 WS 事件类型。hub
|
||
// 订阅这些（并重新归一化为 ws.* 类型），使前端既有的
|
||
// 消费方亮起来，而生产者无需改变其主题名。
|
||
var eventBridge = map[string]string{
|
||
"iot.device_update": "device_update",
|
||
"task.output": "task_output",
|
||
"task.done": "task_done",
|
||
"automation.result": "automation_result",
|
||
"reminder.trigger": "reminder",
|
||
}
|
||
|
||
// bridgePatterns 是 hub 在 ws.> 之外额外订阅的 bus 模式，用于
|
||
// 捕获生产者事件并把它们归一化以供 WS 投递。
|
||
var bridgePatterns = []string{
|
||
"iot.>",
|
||
"task.>",
|
||
"automation.>",
|
||
"reminder.>",
|
||
"cyrene.agent.*.event",
|
||
}
|
||
|
||
// normalizeEvent 把 bus 事件映射为下行 wsEvent。ws.* 主题保持
|
||
// 其设计类型（ws. 前缀被剥掉）；已知的生产者主题被
|
||
// 桥接为其 ws.* 类型；子代理进度事件被塑形为
|
||
// ws.subagent_event。对没有 WS 可见表示的事件返回 ok=false
|
||
// （例如内部的 cyrene.agent ask/reply 流量）。
|
||
func normalizeEvent(e bus.Event) (wsEvent, bool) {
|
||
if typ, ok := eventBridge[e.Subject]; ok {
|
||
return wsEvent{Type: typ, Data: e.Payload}, true
|
||
}
|
||
if strings.HasPrefix(e.Subject, wsPrefix) {
|
||
return wsEvent{Type: eventType(e.Subject), Data: e.Payload}, true
|
||
}
|
||
if ev, ok := subagentProgressEvent(e); ok {
|
||
return ev, true
|
||
}
|
||
return wsEvent{}, false
|
||
}
|
||
|
||
// subagentProgressEvent 把编排进度事件 —— 经 comm.PublishEvent 广播在
|
||
// cyrene.agent.<from>.event 上 —— 映射为 ws.subagent_event
|
||
// 帧。只有五个子代理进度名会被浮现；同一主题上广播的
|
||
// 其他任何事件（ask/reply/addressed 事件，或外来事件
|
||
// 名）都被忽略，使内部 agent 间流量保持内部。
|
||
func subagentProgressEvent(e bus.Event) (wsEvent, bool) {
|
||
if !isBroadcastAgentEvent(e.Subject) {
|
||
return wsEvent{}, false
|
||
}
|
||
ev, ok := e.Payload.(comm.Event)
|
||
if !ok {
|
||
return wsEvent{}, false
|
||
}
|
||
switch ev.Name {
|
||
case orchestration.EventDispatched, orchestration.EventStarted,
|
||
orchestration.EventDone, orchestration.EventFailed, orchestration.EventAborted:
|
||
default:
|
||
return wsEvent{}, false
|
||
}
|
||
p, ok := ev.Data.(orchestration.Progress)
|
||
if !ok {
|
||
return wsEvent{}, false
|
||
}
|
||
return wsEvent{Type: "subagent_event", Data: map[string]any{
|
||
"id": p.AgentID,
|
||
"name": p.AgentID,
|
||
"status": string(p.Status),
|
||
"role": p.RoleID,
|
||
"mode": p.Mode,
|
||
}}, true
|
||
}
|
||
|
||
// streamEventData 把流式的 model.StreamKind 映射为其下游 WS 事件
|
||
// 类型并构建带类型的 data 载荷。它正是把 loop 的结构化
|
||
// 事件变成前端所渲染的 message.*/thinking.*/tool_call.*/tool_execution.*
|
||
// 帧的东西。未知类型回退到旧的 stream_chunk 形状，使
|
||
// 未识别的事件仍能抵达客户端而不是被丢弃。
|
||
func streamEventData(ev model.StreamEvent) (string, map[string]any) {
|
||
switch ev.Kind {
|
||
case model.StreamKindTextStart:
|
||
return "message.start", map[string]any{"index": ev.Index}
|
||
case model.StreamKindTextDelta:
|
||
return "message.delta", map[string]any{"delta": ev.Delta, "index": ev.Index}
|
||
case model.StreamKindTextEnd:
|
||
return "message.end", map[string]any{"index": ev.Index}
|
||
case model.StreamKindThinkingStart:
|
||
return "thinking.start", map[string]any{"index": ev.Index}
|
||
case model.StreamKindThinkingDelta:
|
||
return "thinking.delta", map[string]any{"delta": ev.Delta, "index": ev.Index}
|
||
case model.StreamKindThinkingEnd:
|
||
return "thinking.end", map[string]any{"index": ev.Index}
|
||
case model.StreamKindToolCallStart:
|
||
return "tool_call.start", map[string]any{"id": ev.ID, "index": ev.Index}
|
||
case model.StreamKindToolCallDelta:
|
||
return "tool_call.delta", map[string]any{"id": ev.ID, "index": ev.Index, "delta": ev.Delta}
|
||
case model.StreamKindToolCallEnd:
|
||
return "tool_call.end", map[string]any{"id": ev.ID, "index": ev.Index}
|
||
case model.StreamKindToolExecution:
|
||
return "tool_execution.update", map[string]any{"id": ev.ID, "index": ev.Index, "delta": ev.Delta}
|
||
default:
|
||
return "stream_chunk", map[string]any{"delta": ev.Delta}
|
||
}
|
||
}
|
||
|
||
// isBroadcastAgentEvent 报告 subject 是否恰为 4 段广播形态
|
||
// cyrene.agent.<from>.event，调度器正是这样发布进度的。
|
||
// 它刻意不匹配寻址事件（cyrene.agent.<from>.<to>.event）
|
||
// 或 ask/reply 主题。
|
||
func isBroadcastAgentEvent(subject string) bool {
|
||
parts := strings.Split(subject, ".")
|
||
if len(parts) != 4 {
|
||
return false
|
||
}
|
||
return parts[0] == "cyrene" && parts[1] == "agent" && parts[3] == "event"
|
||
}
|