From 79589b5c9f46d6757284ffe05adb8e802491ab30 Mon Sep 17 00:00:00 2001 From: AskaEth Date: Sat, 3 Oct 2026 15:19:57 +0800 Subject: [PATCH] =?UTF-8?q?refactor(gateway):=20=E6=8B=86=20ws=5Fhub.go=20?= =?UTF-8?q?589=E2=86=92182(=E7=BA=AF=E6=90=AC=E7=A7=BB=E9=9B=B6=E9=80=BB?= =?UTF-8?q?=E8=BE=91)=20=E2=80=94=20conn/turn/origin/message=5Fevent=20?= =?UTF-8?q?=E5=9B=9B=E6=96=87=E4=BB=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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交叉编译绿 --- internal/services/gateway/ws_hub.go | 407 ------------------ internal/services/gateway/ws_hub_conn.go | 165 +++++++ .../services/gateway/ws_hub_message_event.go | 30 ++ internal/services/gateway/ws_hub_origin.go | 73 ++++ internal/services/gateway/ws_hub_turn.go | 168 ++++++++ 5 files changed, 436 insertions(+), 407 deletions(-) create mode 100644 internal/services/gateway/ws_hub_conn.go create mode 100644 internal/services/gateway/ws_hub_message_event.go create mode 100644 internal/services/gateway/ws_hub_origin.go create mode 100644 internal/services/gateway/ws_hub_turn.go diff --git a/internal/services/gateway/ws_hub.go b/internal/services/gateway/ws_hub.go index 2b73642..a507d7d 100644 --- a/internal/services/gateway/ws_hub.go +++ b/internal/services/gateway/ws_hub.go @@ -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, "]"), "[") -} diff --git a/internal/services/gateway/ws_hub_conn.go b/internal/services/gateway/ws_hub_conn.go new file mode 100644 index 0000000..026a767 --- /dev/null +++ b/internal/services/gateway/ws_hub_conn.go @@ -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 "" +} diff --git a/internal/services/gateway/ws_hub_message_event.go b/internal/services/gateway/ws_hub_message_event.go new file mode 100644 index 0000000..586976c --- /dev/null +++ b/internal/services/gateway/ws_hub_message_event.go @@ -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}}) +} diff --git a/internal/services/gateway/ws_hub_origin.go b/internal/services/gateway/ws_hub_origin.go new file mode 100644 index 0000000..dde9df9 --- /dev/null +++ b/internal/services/gateway/ws_hub_origin.go @@ -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, "]"), "[") +} diff --git a/internal/services/gateway/ws_hub_turn.go b/internal/services/gateway/ws_hub_turn.go new file mode 100644 index 0000000..e3a13eb --- /dev/null +++ b/internal/services/gateway/ws_hub_turn.go @@ -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}) +}