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..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..event,调度器正是这样发布进度的。 // 它刻意不匹配寻址事件(cyrene.agent...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" }