Cyrene_v2/internal/services/gateway/ws_hub_turn.go

package gateway

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"encoding/json"
	"strings"
	"time"

	"git.yeij.top/AskaEth/Cyrene_v2/internal/bus"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
)

// relay 是 ws.> 模式与各生产者桥接模式的 bus 处理器。ws.chat 主题是
// 流式触发器（跑 loop 并发 chunk）；ws.* 主题成为其设计类型对应的
// 下行事件；被桥接的生产者主题（device_update、任务 output/done、
// automation result、reminder、subagent progress）被规范化为匹配的
// ws.* 事件类型。没有 WS 可见映射的事件会被丢弃，使内部 agent 间
// 流量永不泄露给客户端。
func (h *hub) relay(ctx context.Context, e bus.Event) error {
	if e.Subject == wsChatSubject {
		go h.streamChat(ctx, e.Payload)
		return nil
	}
	if e.Subject == wsStreamSubject {
		// OnStream→bus→relay 路径：loop 已经发布了结构化的
		// model.StreamEvent；把它映射为带类型的下行事件。
		h.broadcastStream(e.Payload)
		return nil
	}
	ev, ok := normalizeEvent(e)
	if !ok {
		return nil
	}
	h.broadcast(ev)
	return nil
}

// streamChat 是 ws.chat 触发器的入口。它解码请求，并在接了
// orchestrator 时把该轮路由经会话 orchestrator（design/15 的非阻塞
// busy/pending/resume 路径）；否则回退到旧的直接运行方式，使未接
// orchestrator 的 hub 保持完全相同的 stream_chunk/stream_end 契约。
// 它跑在各自的 goroutine 上，因此慢的模型调用不会阻塞 bus 订阅
// worker。
func (h *hub) streamChat(ctx context.Context, payload any) {
	var req chatStreamRequest
	if b, err := json.Marshal(payload); err == nil {
		_ = json.Unmarshal(b, &req)
	}
	if h.runner == nil || (strings.TrimSpace(req.Text) == "" && strings.TrimSpace(req.Image) == "") {
		h.broadcast(wsEvent{Type: "stream_end", TurnID: newTurnID(), Seq: int(h.streamSeq.Add(1)), SessionID: req.SessionID, Data: map[string]any{"stopReason": "idle"}})
		return
	}
	if req.SessionID == "" {
		req.SessionID = "web"
	}
	// Web 主会话是 owner 的私有对话（JWT = owner），因此它绑定全权限
	// env：记忆读写，行为不变。角色来自会话的绑定（design/22 §4.5）：
	// 切到角色 Y 的会话在下一轮以 Y 应答，而未绑定的/既有的会话继续
	// 解析为全局默认角色。
	msgs := userChatMessages(req.Text, req.Image)
	// 通过 ctx 把客户端的 deviceID 与行为模式串进 orchestrator，使
	// 持久化的用户轮次与助手回复都带上标记（design/16 §4.5、§5.1）。
	ctx = withMode(withDeviceID(ctx, req.DeviceID), req.Mode)
	env := h.envResolver.owner(ctx, req.SessionID)
	if h.orchestrator != nil {
		_ = h.orchestrator.Submit(ctx, req.SessionID, msgs, env)
		return
	}
	_, _ = h.runTurn(ctx, req.SessionID, msgs, env)
}

// runTurn 执行一个 agent 轮次并广播其输出流。它是 orchestrator（经其
// 注入的 TurnRunner）与旧回退路径共用的单元，因此无论由谁驱动，一轮
// 总是发出相同的 stream_chunk/stream_end 事件。当 loop 处于流式
// （接了 OnStream）时，结构化增量已经通过 OnStream→bus→relay 路径以
// message.*/thinking.*/tool_call.* 事件扇出，因此只有非流式路径才发出
// 一整段文本 chunk 以维持流契约。
func (h *hub) runTurn(ctx context.Context, sessionID string, msgs []model.Message, env loop.SessionEnv) (*loop.Result, error) {
	turnID := newTurnID()
	// 把轮次身份串进运行上下文，使 loop 把它盖在它发出的每个流式
	// 事件上（message.*/thinking.*/tool_call.*）；hub 随后把它放到
	// 下行帧上。每一轮都取全新的 id —— 一个会话最多一个在途轮次
	// —— 因此客户端能把一串增量关联到同一个助手轮次
	// （design/16 §4.2）。
	ctx = loop.WithTurnID(ctx, turnID)
	if h.runner == nil || len(msgs) == 0 {
		h.broadcast(wsEvent{Type: "stream_end", TurnID: turnID, Seq: int(h.streamSeq.Add(1)), SessionID: sessionID, Data: map[string]any{"stopReason": "idle"}})
		return &loop.Result{StopReason: "idle"}, nil
	}
	res, err := h.runner.Run(loop.WithSessionEnv(ctx, env), sessionID, msgs)
	if err != nil {
		h.log.Error("chat run", "err", err, "session", sessionID)
		h.broadcast(wsEvent{Type: "stream_end", TurnID: turnID, Seq: int(h.streamSeq.Add(1)), SessionID: sessionID, Data: map[string]any{"stopReason": "error", "error": err.Error()}})
		return nil, err
	}
	if !h.streaming {
		h.broadcast(wsEvent{Type: "stream_chunk", TurnID: turnID, Seq: int(h.streamSeq.Add(1)), SessionID: sessionID, Data: map[string]any{"delta": res.Text}})
	}
	h.broadcast(wsEvent{Type: "stream_end", TurnID: turnID, Seq: int(h.streamSeq.Add(1)), SessionID: sessionID, Data: map[string]any{"stopReason": res.StopReason, "steps": res.Steps}})
	return res, nil
}

// handleInterrupt 把客户端的 "interrupt" 帧（design/15 §3）路由到
// orchestrator，取消在途轮次并清空指定会话的待处理队列。未接
// orchestrator 时是空操作。
func (h *hub) handleInterrupt(payload json.RawMessage) {
	if h.orchestrator == nil {
		return
	}
	var req struct {
		SessionID string `json:"sessionId"`
	}
	_ = json.Unmarshal(payload, &req)
	if req.SessionID == "" {
		return
	}
	h.orchestrator.Interrupt(req.SessionID)
}

// broadcastStream 把流式的 model.StreamEvent 映射为其下行 WS 事件类型，
// 并经 hub 的帧门扇出（design/11 §7.1）：每一个结构化流帧都携带模型
// 输出，因此只到达 owner + admin。它跑在 bus 订阅 worker 上；seq 原子
// 分配，因此即便 loop 从多个 goroutine 发出事件，一串帧仍保持有序。
// SessionID/TurnID 随事件本身携带（loop 在轮次边界从运行上下文盖上
// 它们），因此帧带有客户端把它路由到活跃会话以及关联到某个助手
// 轮次所需的范围。
func (h *hub) broadcastStream(payload any) {
	ev, ok := payload.(model.StreamEvent)
	if !ok {
		return
	}
	typ, data := streamEventData(ev)
	h.broadcast(wsEvent{Type: typ, SessionID: ev.SessionID, TurnID: ev.TurnID, Seq: int(h.streamSeq.Add(1)), Data: data})
}

// chatStreamRequest 是解码后的 ws.chat 载荷。
type chatStreamRequest struct {
	SessionID string `json:"sessionId"`
	Text      string `json:"text"`
	Image     string `json:"image,omitempty"`
	// DeviceID 是客户端生成的发送浏览器/设备标识（design/16 §4.5），
	// 例如 "web:chrome:a1b2c3"；它可选，并随消息一同持久化，使 agent
	// 知道主人是从哪一端发消息。
	DeviceID string `json:"deviceId,omitempty"`
	// Mode 是客户端发送该轮时所用的行为模式（design/01 §3、
	// design/16 §5.1）；它可选，按消息持久化，缺省时回退为 "chat"。
	Mode string `json:"mode,omitempty"`
}

// newTurnID 返回新的随机十六进制标识，用于把一串流式 WS 事件关联到
// 同一个助手轮次。它消耗 crypto/rand 源，源不可用时回退到纳秒
// 时间戳，因此总能产出 id（空的 turnId 会静默破坏客户端关联）。
func newTurnID() string {
	var b [8]byte
	if _, err := rand.Read(b[:]); err == nil {
		return "turn-" + hex.EncodeToString(b[:])
	}
	return "turn-" + time.Now().UTC().Format("20060102150405.000000000")
}

// publishChatStream 发布一个 ws.chat 流式触发器（REST 流式模式）。
// deviceID 是发送方的浏览器/设备 id，mode 是客户端发送该轮时所使用的
// 行为模式，二者都随载荷携带，使 ws.chat 消费方能给持久化的轮次打
// 标记（design/16 §4.5、§5.1）。
func (s *Server) publishChatStream(ctx context.Context, sessionID, text, image, deviceID, mode string) {
	if s.bus == nil {
		return
	}
	_ = s.bus.Publish(ctx, wsChatSubject, map[string]any{"sessionId": sessionID, "text": text, "image": image, "deviceId": deviceID, "mode": mode})
}