refactor(gateway): 拆 ws_hub.go 589→182(纯搬移零逻辑) — conn/turn/origin/message_event 四文件
- ws_hub.go 182(hub核心/常量/wsEvent/newHub/broadcast) + ws_hub_conn.go 165(连接层handleWS/read·writePump) + ws_hub_turn.go 168(relay/streamChat/runTurn/interrupt/broadcastStream) + ws_hub_origin.go 73(Origin校验) + ws_hub_message_event.go 30(message_edited/deleted广播) - 代码行多重集 identical + 顶层声明diff空; 每文件≤500/函数≤80(最大handleWS 28/streamChat 25) - gateway测试绿 + build/vet + win/darwin交叉编译绿
This commit is contained in:
AskaEth committed
1 parent
2720d77331
commit
79589b5c9f
5 files changed
+436
-407
No files matched your search
@@ -2,11 +2,8 @@ package gateway
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
@@ -14,12 +11,7 @@ import (
|
||||
|
||||
"github.com/gorilla/websocket"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/auth"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/bus"
|
||||
apperr "git.yeij.top/AskaEth/Cyrene_v2/internal/errors"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||||
)
|
||||
|
||||
// WebSocket tuning constants. The ping/pong cadence keeps intermediary proxies
|
||||
@@ -56,18 +48,6 @@ type wsInbound struct {
|
||||
Data json.RawMessage `json:"data,omitempty"`
|
||||
}
|
||||
|
||||
// wsConn wraps one WebSocket connection. Writes are serialized through the
|
||||
// buffered send channel drained by a single writePump, so the gorilla conn is
|
||||
// only ever touched by one goroutine at a time.
|
||||
type wsConn struct {
|
||||
conn *websocket.Conn
|
||||
send chan []byte
|
||||
done chan struct{}
|
||||
h *hub
|
||||
claims *auth.Claims
|
||||
once sync.Once
|
||||
}
|
||||
|
||||
// hub owns the connection set and relays bus events to every client.
|
||||
type hub struct {
|
||||
mu sync.RWMutex
|
||||
@@ -188,331 +168,11 @@ func (h *hub) broadcast(ev wsEvent) {
|
||||
}
|
||||
}
|
||||
|
||||
// wsMessageEdited / wsMessageDeleted are the downstream event types of the D3
|
||||
// history writes: a message's content was edited, or a message was removed. Both
|
||||
// reuse the wsEvent envelope and the same all-connections fan-out as
|
||||
// `message_persisted` — the hub has no per-session subscription, so the frame
|
||||
// carries SessionID and a client filters it by its active session, exactly as it
|
||||
// does for message_persisted.
|
||||
const (
|
||||
wsMessageEdited = "message_edited"
|
||||
wsMessageDeleted = "message_deleted"
|
||||
)
|
||||
|
||||
// broadcastMessageEdited fans a content edit out to every client as a
|
||||
// `message_edited` frame whose data is the updated message DTO. The frame is a
|
||||
// render-only sync: the edited content's single source of truth stays the history
|
||||
// store, so a client that missed the frame converges again on its next read. The
|
||||
// caller strips Thinking to match the message_persisted/replay contract.
|
||||
func (h *hub) broadcastMessageEdited(msg historyMessage) {
|
||||
h.broadcast(wsEvent{Type: wsMessageEdited, SessionID: msg.SessionID, Data: msg})
|
||||
}
|
||||
|
||||
// broadcastMessageDeleted fans a removal out to every client as a
|
||||
// `message_deleted` frame carrying the removed id (plus the owning sessionId for
|
||||
// client-side session filtering), so every browser of the account drops the row
|
||||
// without a refresh. sessionID may be empty when the store could not report it;
|
||||
// clients then fall back to the globally unique id.
|
||||
func (h *hub) broadcastMessageDeleted(sessionID, id string) {
|
||||
h.broadcast(wsEvent{Type: wsMessageDeleted, SessionID: sessionID, Data: map[string]any{"id": id}})
|
||||
}
|
||||
|
||||
// relay is the bus handler for the ws.> pattern and the producer bridge
|
||||
// patterns. The ws.chat subject is a streaming trigger (run the loop and emit
|
||||
// chunks); a ws.* subject becomes an outbound event of its design type; a
|
||||
// bridged producer subject (device_update, task output/done, automation result,
|
||||
// reminder, subagent progress) is normalized into the matching ws.* event type.
|
||||
// Events with no WS-visible mapping are dropped so internal agent-to-agent
|
||||
// traffic never leaks to clients.
|
||||
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 path: the loop already published a structured
|
||||
// model.StreamEvent; map it to the typed downstream event.
|
||||
h.broadcastStream(e.Payload)
|
||||
return nil
|
||||
}
|
||||
ev, ok := normalizeEvent(e)
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
h.broadcast(ev)
|
||||
return nil
|
||||
}
|
||||
|
||||
// streamChat is the ws.chat trigger entry point. It decodes the request, and
|
||||
// routes the turn through the session orchestrator when one is wired (the
|
||||
// non-blocking busy/pending/resume path of design/15); otherwise it falls back to
|
||||
// the legacy direct run so a hub built without an orchestrator keeps the exact
|
||||
// stream_chunk/stream_end contract. It runs on its own goroutine so a slow model
|
||||
// call does not stall the bus subscription 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"
|
||||
}
|
||||
// The Web main session is the owner's private conversation (JWT = owner), so
|
||||
// it binds the full-access env: memory read+write, unchanged behavior.
|
||||
msgs := userChatMessages(req.Text, req.Image)
|
||||
// Thread the client's deviceID and behavior mode into the orchestrator via ctx
|
||||
// so both the persisted user turn and the assistant reply are tagged
|
||||
// (design/16 §4.5, §5.1).
|
||||
ctx = withMode(withDeviceID(ctx, req.DeviceID), req.Mode)
|
||||
if h.orchestrator != nil {
|
||||
_ = h.orchestrator.Submit(ctx, req.SessionID, msgs, loop.OwnerEnv())
|
||||
return
|
||||
}
|
||||
_, _ = h.runTurn(ctx, req.SessionID, msgs, loop.OwnerEnv())
|
||||
}
|
||||
|
||||
// runTurn executes one agent turn and broadcasts its output stream. It is the
|
||||
// unit both the orchestrator (via its injected TurnRunner) and the legacy
|
||||
// fallback path share, so a turn always emits the same stream_chunk/stream_end
|
||||
// events regardless of who drives it. When the loop is streaming (OnStream wired)
|
||||
// the structured deltas already fan out as message.*/thinking.*/tool_call.*
|
||||
// events through the OnStream→bus→relay path, so only the non-streaming path
|
||||
// emits a single whole-text chunk to hold the stream contract.
|
||||
func (h *hub) runTurn(ctx context.Context, sessionID string, msgs []model.Message, env loop.SessionEnv) (*loop.Result, error) {
|
||||
turnID := newTurnID()
|
||||
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 routes a client "interrupt" frame (design/15 §3) to the
|
||||
// orchestrator, canceling the in-flight turn and clearing the pending queue for
|
||||
// the requested session. It is a no-op when no orchestrator is wired.
|
||||
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 maps a streamed model.StreamEvent to its downstream WS event
|
||||
// type and fans it out to every client. It runs on the bus subscription worker;
|
||||
// seq is assigned atomically so a burst stays ordered even if the loop emits
|
||||
// events from more than one goroutine. TurnId is left empty here because the
|
||||
// OnStream bus payload does not carry the turn identity (future work would
|
||||
// thread a turn id through loop → streamChunk → payload).
|
||||
func (h *hub) broadcastStream(payload any) {
|
||||
ev, ok := payload.(model.StreamEvent)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
typ, data := streamEventData(ev)
|
||||
h.broadcast(wsEvent{Type: typ, Seq: int(h.streamSeq.Add(1)), Data: data})
|
||||
}
|
||||
|
||||
// chatStreamRequest is the decoded ws.chat payload.
|
||||
type chatStreamRequest struct {
|
||||
SessionID string `json:"sessionId"`
|
||||
Text string `json:"text"`
|
||||
Image string `json:"image,omitempty"`
|
||||
// DeviceID is the client-produced identifier of the sending browser/device
|
||||
// (design/16 §4.5), e.g. "web:chrome:a1b2c3"; it is optional and persisted
|
||||
// alongside the message so the agent knows which end the owner is chatting from.
|
||||
DeviceID string `json:"deviceId,omitempty"`
|
||||
// Mode is the behavior mode the client sent the turn in (design/01 §3,
|
||||
// design/16 §5.1); it is optional and persisted per message, falling back to
|
||||
// "chat" when omitted.
|
||||
Mode string `json:"mode,omitempty"`
|
||||
}
|
||||
|
||||
// handleWS upgrades /ws/chat after authenticating the JWT from the query token or
|
||||
// the Authorization header. A failed upgrade writes a 401 {error:...} before the
|
||||
// handshake, so a browser/IM client sees a clean reject.
|
||||
func (s *Server) handleWS(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Method != http.MethodGet {
|
||||
writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "websocket requires GET"))
|
||||
return
|
||||
}
|
||||
claims, err := s.auth.Verify(wsToken(r))
|
||||
if err != nil {
|
||||
s.log.Debug("ws auth", "err", err)
|
||||
writeError(w, apperr.ErrUnauthorized)
|
||||
return
|
||||
}
|
||||
conn, err := s.hub.upgrader.Upgrade(w, r, nil)
|
||||
if err != nil {
|
||||
s.log.Debug("ws upgrade", "err", err)
|
||||
return
|
||||
}
|
||||
c := &wsConn{
|
||||
conn: conn,
|
||||
send: make(chan []byte, wsSendBuffer),
|
||||
done: make(chan struct{}),
|
||||
h: s.hub,
|
||||
claims: claims,
|
||||
}
|
||||
s.hub.register(c)
|
||||
s.hub.replayHistory(c, history.MainSessionID)
|
||||
go c.writePump()
|
||||
go c.readPump()
|
||||
}
|
||||
|
||||
// readPump reads client frames until the connection closes, routing each by type.
|
||||
func (c *wsConn) readPump() {
|
||||
defer func() {
|
||||
c.h.unregister(c)
|
||||
c.shutdown()
|
||||
}()
|
||||
c.conn.SetReadLimit(wsReadLimit)
|
||||
_ = c.conn.SetReadDeadline(time.Now().Add(wsPongWait))
|
||||
c.conn.SetPongHandler(func(string) error {
|
||||
return c.conn.SetReadDeadline(time.Now().Add(wsPongWait))
|
||||
})
|
||||
for {
|
||||
_, msg, err := c.conn.ReadMessage()
|
||||
if err != nil {
|
||||
if websocket.IsUnexpectedCloseError(err, websocket.CloseNormalClosure, websocket.CloseGoingAway) {
|
||||
c.h.log.Debug("ws read", "err", err)
|
||||
}
|
||||
return
|
||||
}
|
||||
var in wsInbound
|
||||
if err := json.Unmarshal(msg, &in); err != nil {
|
||||
c.enqueueJSON(wsEvent{Type: "error", Data: map[string]any{"message": "invalid json"}})
|
||||
continue
|
||||
}
|
||||
c.route(in)
|
||||
}
|
||||
}
|
||||
|
||||
// writePump drains the send channel (and sends keepalive pings) until the
|
||||
// connection closes. It is the only writer on the gorilla conn.
|
||||
func (c *wsConn) writePump() {
|
||||
ticker := time.NewTicker(wsPingPeriod)
|
||||
defer func() {
|
||||
ticker.Stop()
|
||||
_ = c.conn.Close()
|
||||
}()
|
||||
for {
|
||||
select {
|
||||
case <-c.done:
|
||||
return
|
||||
case msg := <-c.send:
|
||||
_ = c.conn.SetWriteDeadline(time.Now().Add(wsWriteWait))
|
||||
if err := c.conn.WriteMessage(websocket.TextMessage, msg); err != nil {
|
||||
return
|
||||
}
|
||||
case <-ticker.C:
|
||||
_ = c.conn.SetWriteDeadline(time.Now().Add(wsWriteWait))
|
||||
if err := c.conn.WriteMessage(websocket.PingMessage, nil); err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// publishPublic publishes a bus event without any per-connection reply.
|
||||
func (c *wsConn) publishPublic(subject string, payload any) {
|
||||
if c.h.bus == nil {
|
||||
return
|
||||
}
|
||||
_ = c.h.bus.Publish(context.Background(), subject, payload)
|
||||
}
|
||||
|
||||
// shutdown closes the connection exactly once. It tolerates a nil *websocket.Conn
|
||||
// so tests that build a fake connection can rely on a safe teardown.
|
||||
func (c *wsConn) shutdown() {
|
||||
c.once.Do(func() {
|
||||
close(c.done)
|
||||
if c.conn != nil {
|
||||
_ = c.conn.Close()
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// enqueue queues a serialized frame, dropping it when a slow client's buffer is
|
||||
// full so the producer never blocks.
|
||||
func (c *wsConn) enqueue(b []byte) {
|
||||
select {
|
||||
case c.send <- b:
|
||||
default:
|
||||
}
|
||||
}
|
||||
|
||||
// enqueueJSON serializes ev and queues it.
|
||||
func (c *wsConn) enqueueJSON(ev wsEvent) {
|
||||
b, err := json.Marshal(ev)
|
||||
if err != nil {
|
||||
c.h.log.Warn("marshal ws event", "err", err)
|
||||
return
|
||||
}
|
||||
c.enqueue(b)
|
||||
}
|
||||
|
||||
// publishChatStream publishes a ws.chat streaming trigger (REST stream mode).
|
||||
// deviceID is the sending browser/device id and mode the behavior mode the client
|
||||
// sent the turn in, both carried on the payload so the ws.chat consumer can tag
|
||||
// the persisted turn (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})
|
||||
}
|
||||
|
||||
// wsToken extracts the JWT from the ?token= query or the Authorization header.
|
||||
func wsToken(r *http.Request) string {
|
||||
if t := r.URL.Query().Get("token"); t != "" {
|
||||
return t
|
||||
}
|
||||
const prefix = "Bearer "
|
||||
h := r.Header.Get("Authorization")
|
||||
if strings.HasPrefix(h, prefix) {
|
||||
return strings.TrimSpace(h[len(prefix):])
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
// eventType maps a bus subject to its outbound event type, dropping the ws. prefix.
|
||||
func eventType(subject string) string {
|
||||
return strings.TrimPrefix(subject, wsPrefix)
|
||||
}
|
||||
|
||||
// newTurnID returns a fresh random hex identifier used to correlate a burst of
|
||||
// streamed WS events to one assistant turn. It burns the crypto/rand source and
|
||||
// falls back to a nanosecond timestamp if the source is unavailable, so a id is
|
||||
// always produced (a blank turnId would break client correlation silently).
|
||||
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")
|
||||
}
|
||||
|
||||
// rawData converts a possibly-empty raw payload into a value safe to publish.
|
||||
func rawData(d json.RawMessage) any {
|
||||
if len(d) == 0 {
|
||||
@@ -520,70 +180,3 @@ func rawData(d json.RawMessage) any {
|
||||
}
|
||||
return d
|
||||
}
|
||||
|
||||
// originChecker allows same-origin, same-host, and whitelisted WebSocket origins,
|
||||
// mirroring the CORS allow list. A "*" allows all.
|
||||
//
|
||||
// Same-host relaxation: when cyrene serves its own embedded frontend (production
|
||||
// mode), the page origin is exactly the gateway's Host (e.g. http://localhost:10942),
|
||||
// which is not in the dev-phase ALLOWED_ORIGINS list. We must let that through so a
|
||||
// user who changes PORT does not need to edit .env for the WS to connect. Only truly
|
||||
// cross-origin clients (external IM / dashboards on another origin) hit the allow-list.
|
||||
func originChecker(allowed map[string]bool, allowAll bool) func(*http.Request) bool {
|
||||
return func(r *http.Request) bool {
|
||||
origin := r.Header.Get("Origin")
|
||||
if origin == "" {
|
||||
return true
|
||||
}
|
||||
if allowAll {
|
||||
return true
|
||||
}
|
||||
if allowed[origin] {
|
||||
return true
|
||||
}
|
||||
// Same-host: the Origin matches this request's Host (scheme://host:port == host[:port]).
|
||||
return sameOriginAsHost(origin, r)
|
||||
}
|
||||
}
|
||||
|
||||
// sameOriginAsHost reports whether origin (a full URL origin string such as
|
||||
// "http://localhost:10942") corresponds to this request's Host header. It compares
|
||||
// the host[:port] portion with loopback equivalence: "localhost" and "127.0.0.1"
|
||||
// (and the IPv6 form) are treated as the same host, so serving the embedded frontend
|
||||
// on 127.0.0.1 while a client connects via localhost (or vice versa) still passes.
|
||||
func sameOriginAsHost(origin string, r *http.Request) bool {
|
||||
if origin == "" || r == nil {
|
||||
return false
|
||||
}
|
||||
after := origin
|
||||
if i := strings.Index(after, "://"); i >= 0 {
|
||||
after = after[i+3:]
|
||||
}
|
||||
if i := strings.IndexByte(after, '/'); i >= 0 {
|
||||
after = after[:i]
|
||||
}
|
||||
return hostEquivalent(after, r.Host)
|
||||
}
|
||||
|
||||
// hostEquivalent compares two host[:port] strings, treating empty-port variants,
|
||||
// case, and loopback aliases as equal. It handles "localhost" vs "127.0.0.1" and
|
||||
// bracketed IPv6 loopback "[::1]".
|
||||
func hostEquivalent(a, b string) bool {
|
||||
na := normalizeHost(a)
|
||||
nb := normalizeHost(b)
|
||||
if na == nb {
|
||||
return true
|
||||
}
|
||||
// Loopback aliases: localhost, 127.0.0.1, [::1]
|
||||
loop := map[string]bool{"localhost": true, "127.0.0.1": true, "[::1]": true, "::1": true, "127.0.0.2": true}
|
||||
return loop[na] && loop[nb]
|
||||
}
|
||||
|
||||
// normalizeHost strips the port (if present), lowercases, and trims brackets.
|
||||
func normalizeHost(h string) string {
|
||||
h = strings.ToLower(strings.TrimSpace(h))
|
||||
if i := strings.LastIndex(h, ":"); i >= 0 && !strings.Contains(h[i:], "]") {
|
||||
h = h[:i]
|
||||
}
|
||||
return strings.TrimPrefix(strings.TrimSuffix(h, "]"), "[")
|
||||
}
|
||||
@@ -0,0 +1,165 @@
|
||||
package gateway
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/gorilla/websocket"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/auth"
|
||||
apperr "git.yeij.top/AskaEth/Cyrene_v2/internal/errors"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history"
|
||||
)
|
||||
|
||||
// wsConn wraps one WebSocket connection. Writes are serialized through the
|
||||
// buffered send channel drained by a single writePump, so the gorilla conn is
|
||||
// only ever touched by one goroutine at a time.
|
||||
type wsConn struct {
|
||||
conn *websocket.Conn
|
||||
send chan []byte
|
||||
done chan struct{}
|
||||
h *hub
|
||||
claims *auth.Claims
|
||||
once sync.Once
|
||||
}
|
||||
|
||||
// handleWS upgrades /ws/chat after authenticating the JWT from the query token or
|
||||
// the Authorization header. A failed upgrade writes a 401 {error:...} before the
|
||||
// handshake, so a browser/IM client sees a clean reject.
|
||||
func (s *Server) handleWS(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Method != http.MethodGet {
|
||||
writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "websocket requires GET"))
|
||||
return
|
||||
}
|
||||
claims, err := s.auth.Verify(wsToken(r))
|
||||
if err != nil {
|
||||
s.log.Debug("ws auth", "err", err)
|
||||
writeError(w, apperr.ErrUnauthorized)
|
||||
return
|
||||
}
|
||||
conn, err := s.hub.upgrader.Upgrade(w, r, nil)
|
||||
if err != nil {
|
||||
s.log.Debug("ws upgrade", "err", err)
|
||||
return
|
||||
}
|
||||
c := &wsConn{
|
||||
conn: conn,
|
||||
send: make(chan []byte, wsSendBuffer),
|
||||
done: make(chan struct{}),
|
||||
h: s.hub,
|
||||
claims: claims,
|
||||
}
|
||||
s.hub.register(c)
|
||||
s.hub.replayHistory(c, history.MainSessionID)
|
||||
go c.writePump()
|
||||
go c.readPump()
|
||||
}
|
||||
|
||||
// readPump reads client frames until the connection closes, routing each by type.
|
||||
func (c *wsConn) readPump() {
|
||||
defer func() {
|
||||
c.h.unregister(c)
|
||||
c.shutdown()
|
||||
}()
|
||||
c.conn.SetReadLimit(wsReadLimit)
|
||||
_ = c.conn.SetReadDeadline(time.Now().Add(wsPongWait))
|
||||
c.conn.SetPongHandler(func(string) error {
|
||||
return c.conn.SetReadDeadline(time.Now().Add(wsPongWait))
|
||||
})
|
||||
for {
|
||||
_, msg, err := c.conn.ReadMessage()
|
||||
if err != nil {
|
||||
if websocket.IsUnexpectedCloseError(err, websocket.CloseNormalClosure, websocket.CloseGoingAway) {
|
||||
c.h.log.Debug("ws read", "err", err)
|
||||
}
|
||||
return
|
||||
}
|
||||
var in wsInbound
|
||||
if err := json.Unmarshal(msg, &in); err != nil {
|
||||
c.enqueueJSON(wsEvent{Type: "error", Data: map[string]any{"message": "invalid json"}})
|
||||
continue
|
||||
}
|
||||
c.route(in)
|
||||
}
|
||||
}
|
||||
|
||||
// writePump drains the send channel (and sends keepalive pings) until the
|
||||
// connection closes. It is the only writer on the gorilla conn.
|
||||
func (c *wsConn) writePump() {
|
||||
ticker := time.NewTicker(wsPingPeriod)
|
||||
defer func() {
|
||||
ticker.Stop()
|
||||
_ = c.conn.Close()
|
||||
}()
|
||||
for {
|
||||
select {
|
||||
case <-c.done:
|
||||
return
|
||||
case msg := <-c.send:
|
||||
_ = c.conn.SetWriteDeadline(time.Now().Add(wsWriteWait))
|
||||
if err := c.conn.WriteMessage(websocket.TextMessage, msg); err != nil {
|
||||
return
|
||||
}
|
||||
case <-ticker.C:
|
||||
_ = c.conn.SetWriteDeadline(time.Now().Add(wsWriteWait))
|
||||
if err := c.conn.WriteMessage(websocket.PingMessage, nil); err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// publishPublic publishes a bus event without any per-connection reply.
|
||||
func (c *wsConn) publishPublic(subject string, payload any) {
|
||||
if c.h.bus == nil {
|
||||
return
|
||||
}
|
||||
_ = c.h.bus.Publish(context.Background(), subject, payload)
|
||||
}
|
||||
|
||||
// shutdown closes the connection exactly once. It tolerates a nil *websocket.Conn
|
||||
// so tests that build a fake connection can rely on a safe teardown.
|
||||
func (c *wsConn) shutdown() {
|
||||
c.once.Do(func() {
|
||||
close(c.done)
|
||||
if c.conn != nil {
|
||||
_ = c.conn.Close()
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// enqueue queues a serialized frame, dropping it when a slow client's buffer is
|
||||
// full so the producer never blocks.
|
||||
func (c *wsConn) enqueue(b []byte) {
|
||||
select {
|
||||
case c.send <- b:
|
||||
default:
|
||||
}
|
||||
}
|
||||
|
||||
// enqueueJSON serializes ev and queues it.
|
||||
func (c *wsConn) enqueueJSON(ev wsEvent) {
|
||||
b, err := json.Marshal(ev)
|
||||
if err != nil {
|
||||
c.h.log.Warn("marshal ws event", "err", err)
|
||||
return
|
||||
}
|
||||
c.enqueue(b)
|
||||
}
|
||||
|
||||
// wsToken extracts the JWT from the ?token= query or the Authorization header.
|
||||
func wsToken(r *http.Request) string {
|
||||
if t := r.URL.Query().Get("token"); t != "" {
|
||||
return t
|
||||
}
|
||||
const prefix = "Bearer "
|
||||
h := r.Header.Get("Authorization")
|
||||
if strings.HasPrefix(h, prefix) {
|
||||
return strings.TrimSpace(h[len(prefix):])
|
||||
}
|
||||
return ""
|
||||
}
|
||||
@@ -0,0 +1,30 @@
|
||||
package gateway
|
||||
|
||||
// wsMessageEdited / wsMessageDeleted are the downstream event types of the D3
|
||||
// history writes: a message's content was edited, or a message was removed. Both
|
||||
// reuse the wsEvent envelope and the same all-connections fan-out as
|
||||
// `message_persisted` — the hub has no per-session subscription, so the frame
|
||||
// carries SessionID and a client filters it by its active session, exactly as it
|
||||
// does for message_persisted.
|
||||
const (
|
||||
wsMessageEdited = "message_edited"
|
||||
wsMessageDeleted = "message_deleted"
|
||||
)
|
||||
|
||||
// broadcastMessageEdited fans a content edit out to every client as a
|
||||
// `message_edited` frame whose data is the updated message DTO. The frame is a
|
||||
// render-only sync: the edited content's single source of truth stays the history
|
||||
// store, so a client that missed the frame converges again on its next read. The
|
||||
// caller strips Thinking to match the message_persisted/replay contract.
|
||||
func (h *hub) broadcastMessageEdited(msg historyMessage) {
|
||||
h.broadcast(wsEvent{Type: wsMessageEdited, SessionID: msg.SessionID, Data: msg})
|
||||
}
|
||||
|
||||
// broadcastMessageDeleted fans a removal out to every client as a
|
||||
// `message_deleted` frame carrying the removed id (plus the owning sessionId for
|
||||
// client-side session filtering), so every browser of the account drops the row
|
||||
// without a refresh. sessionID may be empty when the store could not report it;
|
||||
// clients then fall back to the globally unique id.
|
||||
func (h *hub) broadcastMessageDeleted(sessionID, id string) {
|
||||
h.broadcast(wsEvent{Type: wsMessageDeleted, SessionID: sessionID, Data: map[string]any{"id": id}})
|
||||
}
|
||||
@@ -0,0 +1,73 @@
|
||||
package gateway
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// originChecker allows same-origin, same-host, and whitelisted WebSocket origins,
|
||||
// mirroring the CORS allow list. A "*" allows all.
|
||||
//
|
||||
// Same-host relaxation: when cyrene serves its own embedded frontend (production
|
||||
// mode), the page origin is exactly the gateway's Host (e.g. http://localhost:10942),
|
||||
// which is not in the dev-phase ALLOWED_ORIGINS list. We must let that through so a
|
||||
// user who changes PORT does not need to edit .env for the WS to connect. Only truly
|
||||
// cross-origin clients (external IM / dashboards on another origin) hit the allow-list.
|
||||
func originChecker(allowed map[string]bool, allowAll bool) func(*http.Request) bool {
|
||||
return func(r *http.Request) bool {
|
||||
origin := r.Header.Get("Origin")
|
||||
if origin == "" {
|
||||
return true
|
||||
}
|
||||
if allowAll {
|
||||
return true
|
||||
}
|
||||
if allowed[origin] {
|
||||
return true
|
||||
}
|
||||
// Same-host: the Origin matches this request's Host (scheme://host:port == host[:port]).
|
||||
return sameOriginAsHost(origin, r)
|
||||
}
|
||||
}
|
||||
|
||||
// sameOriginAsHost reports whether origin (a full URL origin string such as
|
||||
// "http://localhost:10942") corresponds to this request's Host header. It compares
|
||||
// the host[:port] portion with loopback equivalence: "localhost" and "127.0.0.1"
|
||||
// (and the IPv6 form) are treated as the same host, so serving the embedded frontend
|
||||
// on 127.0.0.1 while a client connects via localhost (or vice versa) still passes.
|
||||
func sameOriginAsHost(origin string, r *http.Request) bool {
|
||||
if origin == "" || r == nil {
|
||||
return false
|
||||
}
|
||||
after := origin
|
||||
if i := strings.Index(after, "://"); i >= 0 {
|
||||
after = after[i+3:]
|
||||
}
|
||||
if i := strings.IndexByte(after, '/'); i >= 0 {
|
||||
after = after[:i]
|
||||
}
|
||||
return hostEquivalent(after, r.Host)
|
||||
}
|
||||
|
||||
// hostEquivalent compares two host[:port] strings, treating empty-port variants,
|
||||
// case, and loopback aliases as equal. It handles "localhost" vs "127.0.0.1" and
|
||||
// bracketed IPv6 loopback "[::1]".
|
||||
func hostEquivalent(a, b string) bool {
|
||||
na := normalizeHost(a)
|
||||
nb := normalizeHost(b)
|
||||
if na == nb {
|
||||
return true
|
||||
}
|
||||
// Loopback aliases: localhost, 127.0.0.1, [::1]
|
||||
loop := map[string]bool{"localhost": true, "127.0.0.1": true, "[::1]": true, "::1": true, "127.0.0.2": true}
|
||||
return loop[na] && loop[nb]
|
||||
}
|
||||
|
||||
// normalizeHost strips the port (if present), lowercases, and trims brackets.
|
||||
func normalizeHost(h string) string {
|
||||
h = strings.ToLower(strings.TrimSpace(h))
|
||||
if i := strings.LastIndex(h, ":"); i >= 0 && !strings.Contains(h[i:], "]") {
|
||||
h = h[:i]
|
||||
}
|
||||
return strings.TrimPrefix(strings.TrimSuffix(h, "]"), "[")
|
||||
}
|
||||
@@ -0,0 +1,168 @@
|
||||
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 is the bus handler for the ws.> pattern and the producer bridge
|
||||
// patterns. The ws.chat subject is a streaming trigger (run the loop and emit
|
||||
// chunks); a ws.* subject becomes an outbound event of its design type; a
|
||||
// bridged producer subject (device_update, task output/done, automation result,
|
||||
// reminder, subagent progress) is normalized into the matching ws.* event type.
|
||||
// Events with no WS-visible mapping are dropped so internal agent-to-agent
|
||||
// traffic never leaks to clients.
|
||||
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 path: the loop already published a structured
|
||||
// model.StreamEvent; map it to the typed downstream event.
|
||||
h.broadcastStream(e.Payload)
|
||||
return nil
|
||||
}
|
||||
ev, ok := normalizeEvent(e)
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
h.broadcast(ev)
|
||||
return nil
|
||||
}
|
||||
|
||||
// streamChat is the ws.chat trigger entry point. It decodes the request, and
|
||||
// routes the turn through the session orchestrator when one is wired (the
|
||||
// non-blocking busy/pending/resume path of design/15); otherwise it falls back to
|
||||
// the legacy direct run so a hub built without an orchestrator keeps the exact
|
||||
// stream_chunk/stream_end contract. It runs on its own goroutine so a slow model
|
||||
// call does not stall the bus subscription 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"
|
||||
}
|
||||
// The Web main session is the owner's private conversation (JWT = owner), so
|
||||
// it binds the full-access env: memory read+write, unchanged behavior.
|
||||
msgs := userChatMessages(req.Text, req.Image)
|
||||
// Thread the client's deviceID and behavior mode into the orchestrator via ctx
|
||||
// so both the persisted user turn and the assistant reply are tagged
|
||||
// (design/16 §4.5, §5.1).
|
||||
ctx = withMode(withDeviceID(ctx, req.DeviceID), req.Mode)
|
||||
if h.orchestrator != nil {
|
||||
_ = h.orchestrator.Submit(ctx, req.SessionID, msgs, loop.OwnerEnv())
|
||||
return
|
||||
}
|
||||
_, _ = h.runTurn(ctx, req.SessionID, msgs, loop.OwnerEnv())
|
||||
}
|
||||
|
||||
// runTurn executes one agent turn and broadcasts its output stream. It is the
|
||||
// unit both the orchestrator (via its injected TurnRunner) and the legacy
|
||||
// fallback path share, so a turn always emits the same stream_chunk/stream_end
|
||||
// events regardless of who drives it. When the loop is streaming (OnStream wired)
|
||||
// the structured deltas already fan out as message.*/thinking.*/tool_call.*
|
||||
// events through the OnStream→bus→relay path, so only the non-streaming path
|
||||
// emits a single whole-text chunk to hold the stream contract.
|
||||
func (h *hub) runTurn(ctx context.Context, sessionID string, msgs []model.Message, env loop.SessionEnv) (*loop.Result, error) {
|
||||
turnID := newTurnID()
|
||||
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 routes a client "interrupt" frame (design/15 §3) to the
|
||||
// orchestrator, canceling the in-flight turn and clearing the pending queue for
|
||||
// the requested session. It is a no-op when no orchestrator is wired.
|
||||
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 maps a streamed model.StreamEvent to its downstream WS event
|
||||
// type and fans it out to every client. It runs on the bus subscription worker;
|
||||
// seq is assigned atomically so a burst stays ordered even if the loop emits
|
||||
// events from more than one goroutine. TurnId is left empty here because the
|
||||
// OnStream bus payload does not carry the turn identity (future work would
|
||||
// thread a turn id through loop → streamChunk → payload).
|
||||
func (h *hub) broadcastStream(payload any) {
|
||||
ev, ok := payload.(model.StreamEvent)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
typ, data := streamEventData(ev)
|
||||
h.broadcast(wsEvent{Type: typ, Seq: int(h.streamSeq.Add(1)), Data: data})
|
||||
}
|
||||
|
||||
// chatStreamRequest is the decoded ws.chat payload.
|
||||
type chatStreamRequest struct {
|
||||
SessionID string `json:"sessionId"`
|
||||
Text string `json:"text"`
|
||||
Image string `json:"image,omitempty"`
|
||||
// DeviceID is the client-produced identifier of the sending browser/device
|
||||
// (design/16 §4.5), e.g. "web:chrome:a1b2c3"; it is optional and persisted
|
||||
// alongside the message so the agent knows which end the owner is chatting from.
|
||||
DeviceID string `json:"deviceId,omitempty"`
|
||||
// Mode is the behavior mode the client sent the turn in (design/01 §3,
|
||||
// design/16 §5.1); it is optional and persisted per message, falling back to
|
||||
// "chat" when omitted.
|
||||
Mode string `json:"mode,omitempty"`
|
||||
}
|
||||
|
||||
// newTurnID returns a fresh random hex identifier used to correlate a burst of
|
||||
// streamed WS events to one assistant turn. It burns the crypto/rand source and
|
||||
// falls back to a nanosecond timestamp if the source is unavailable, so a id is
|
||||
// always produced (a blank turnId would break client correlation silently).
|
||||
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 publishes a ws.chat streaming trigger (REST stream mode).
|
||||
// deviceID is the sending browser/device id and mode the behavior mode the client
|
||||
// sent the turn in, both carried on the payload so the ws.chat consumer can tag
|
||||
// the persisted turn (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})
|
||||
}
|
||||
Reference in new issue
Block a user