Cyrene_v2/internal/services/gateway/ws_bridge.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.*
// 与 turn.start/end 帧的东西。模型流帧（message.*/thinking.*/tool_call.*）
// 额外携带 streamId（同一模型流的多帧关联键，design/04 §1.1）；
// tool_execution.* 与 turn.* 不属于某一次模型流，不带 streamId。
// 未知类型回退到旧的 stream_chunk 形状，使
// 未识别的事件仍能抵达客户端而不是被丢弃。
func streamEventData(ev model.StreamEvent) (string, map[string]any) {
	switch ev.Kind {
	case model.StreamKindTextStart:
		return "message.start", withStreamID(map[string]any{"index": ev.Index}, ev)
	case model.StreamKindTextDelta:
		return "message.delta", withStreamID(map[string]any{"delta": ev.Delta, "index": ev.Index}, ev)
	case model.StreamKindTextEnd:
		return "message.end", withStreamID(map[string]any{"index": ev.Index}, ev)
	case model.StreamKindThinkingStart:
		return "thinking.start", withStreamID(map[string]any{"index": ev.Index}, ev)
	case model.StreamKindThinkingDelta:
		return "thinking.delta", withStreamID(map[string]any{"delta": ev.Delta, "index": ev.Index}, ev)
	case model.StreamKindThinkingEnd:
		return "thinking.end", withStreamID(map[string]any{"index": ev.Index}, ev)
	case model.StreamKindToolCallStart:
		return "tool_call.start", withStreamID(map[string]any{"id": ev.ID, "index": ev.Index}, ev)
	case model.StreamKindToolCallDelta:
		return "tool_call.delta", withStreamID(map[string]any{"id": ev.ID, "index": ev.Index, "delta": ev.Delta}, ev)
	case model.StreamKindToolCallEnd:
		return "tool_call.end", withStreamID(map[string]any{"id": ev.ID, "index": ev.Index}, ev)
	case model.StreamKindToolExecutionStart:
		return "tool_execution.start", toolExecutionData(ev)
	case model.StreamKindToolExecutionUpdate:
		return "tool_execution.update", toolExecutionData(ev)
	case model.StreamKindToolExecutionEnd:
		return "tool_execution.end", toolExecutionData(ev)
	case model.StreamKindTurnStart:
		return "turn.start", nil
	case model.StreamKindTurnEnd:
		return "turn.end", nil
	default:
		return "stream_chunk", map[string]any{"delta": ev.Delta}
	}
}

// withStreamID 把事件携带的 streamId 并入模型流帧的 data（空值省略，
// 使未带流身份的事件保持旧形状）。它原地修改并返回 data，
// 仅供 streamEventData 的字面量使用。
func withStreamID(data map[string]any, ev model.StreamEvent) map[string]any {
	if ev.StreamID != "" {
		data["streamId"] = ev.StreamID
	}
	return data
}

// toolExecutionData 构建 tool_execution.* 帧的载荷：工具名与细粒度
// phase 为结构化字段（phase 绝不塞进 delta，design/04 §1.1 裁决
// 01.4.1）；id 复用 provider 的 tool call id（可关联 tool_call.* 帧），
// update 帧可再携带 delta 局部输出。键在空值时省略。
func toolExecutionData(ev model.StreamEvent) map[string]any {
	data := map[string]any{"tool": ev.Tool, "phase": ev.Phase}
	if ev.ID != "" {
		data["id"] = ev.ID
	}
	if ev.Kind == model.StreamKindToolExecutionUpdate && ev.Delta != "" {
		data["delta"] = ev.Delta
	}
	return data
}

// 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"
}